Python之FastAPI 开发知识(第五篇)

在前面的教程中，我们深入探讨了 FastAPI 的一系列高级特性，这些特性使得 FastAPI 不仅是一个简单的 API 框架，更是一个能够构建复杂、高性能、生产级应用的全能工具。以下是核心高级特性的简要回顾：

1. 性能优化

• 异步数据库操作
  ：使用异步驱动（如 asyncpg）配合 SQLAlchemy 异步扩展，实现真正的非阻塞 I/O。
• 多级缓存
  ：内存缓存（cachetools）与分布式缓存（Redis）相结合，降低数据库负载。
• Gzip 压缩
  ：通过 GZipMiddleware 减少网络传输量。
• 速率限制
  ：使用 slowapi 防止滥用。
• 连接池优化
  ：合理配置数据库连接池参数，平衡资源与性能。
• 性能监控
  ：中间件记录请求耗时，定位瓶颈。

2. 高级安全实践

• 多因素认证 (MFA)
  ：集成 TOTP（如 Google Authenticator）增强账户安全。
• API 密钥管理
  ：为第三方开发者提供可过期、带权限的 API 密钥。
• 增强的密码哈希
  ：使用 passlib 配置多种哈希算法（bcrypt、argon2），支持密码升级。
• CSRF 保护
  ：对于非 API 的传统表单，添加 CSRF 令牌验证。

3. 异步任务处理

• BackgroundTasks
  ：轻量级后台任务，适用于简单操作。
• Celery 集成
  ：分布式任务队列，支持重试、定时、任务状态跟踪。
• APScheduler
  ：实现定时任务，如定期清理过期数据。

4. 测试驱动开发

• 单元测试
  ：使用 pytest 和 TestClient 编写测试，覆盖各种场景。
• 依赖项覆盖
  ：通过 app.dependency_overrides 替换真实依赖（如数据库）。
• 异步测试
  ：使用 AsyncClient 测试异步端点。
• 测试覆盖率
  ：pytest-cov 测量覆盖率。

5. GraphQL 集成

• Strawberry
  ：将 FastAPI 与 GraphQL 结合，提供灵活的数据查询。
• Query 与 Mutation
  ：定义 GraphQL 类型和操作，与 REST 共存。

6. 微服务架构

• 服务发现
  ：使用 Consul 注册与发现服务实例。
• 服务间通信
  ：httpx.AsyncClient 实现异步 HTTP 调用。
• 容错与重试
  ：结合 tenacity 等库增强健壮性。

这些高级特性相互配合，能够帮助你构建出高可用、可扩展、安全的现代 Web 应用。下面，我们将通过一个完整的实战项目——“实时聊天 + 用户认证”——来演示如何将这些知识付诸实践。

完整项目：实时聊天 + 用户认证

本项目实现了一个简单的聊天应用，包含以下功能：

• 用户注册、登录（JWT 认证）
• 受保护的用户信息接口
• WebSocket 聊天室，支持多用户实时广播消息
• WebSocket 连接需携带 JWT 令牌进行身份验证

项目结构

chat_app/
├── main.py                 # 应用入口
├── database.py             # 数据库配置
├── models.py               # SQLAlchemy 模型
├── schemas.py              # Pydantic 模型
├── auth.py                 # 认证相关（密码哈希、JWT）
├── dependencies.py         # 公共依赖项
├── routers/
│   ├── auth.py             # 认证路由
│   ├── users.py            # 用户信息路由
│   └── chat.py             # 聊天 WebSocket 路由
├── requirements.txt        # 依赖列表
└── README.md               # 项目说明

环境要求

• Python 3.8+
• 安装依赖：pip install -r requirements.txt

依赖清单 (requirements.txt)

fastapi
uvicorn[standard]
sqlalchemy
python-jose[cryptography]
passlib[bcrypt]
python-multipart

代码实现

1. database.py - 数据库配置

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL ="sqlite:///./chat_app.db"
# SQLite 需要设置 check_same_thread=False
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread":False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

2. models.py - SQLAlchemy 模型

from sqlalchemy import Column, Integer, String
from.database import Base

classUser(Base):
    __tablename__ ="users"

id= Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)

3. schemas.py - Pydantic 模型

from pydantic import BaseModel

classUserBase(BaseModel):
    username:str

classUserCreate(UserBase):
    password:str

classUser(UserBase):
id:int
classConfig:
        orm_mode =True

classToken(BaseModel):
    access_token:str
    token_type:str

classTokenData(BaseModel):
    username:str|None=None

4. auth.py - 认证工具

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext

# 密钥应从环境变量读取，此处仅作示例
SECRET_KEY ="09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM ="HS256"
ACCESS_TOKEN_EXPIRE_MINUTES =30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

defverify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)

defget_password_hash(password):
return pwd_context.hash(password)

defcreate_access_token(data:dict, expires_delta: timedelta |None=None):
    to_encode = data.copy()
if expires_delta:
        expire = datetime.utcnow()+ expires_delta
else:
        expire = datetime.utcnow()+ timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt

defdecode_token(token:str):
try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username:str= payload.get("sub")
if username isNone:
returnNone
return username
except JWTError:
returnNone

5. dependencies.py - 公共依赖项

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from jose import JWTError

from.import database, models, auth, schemas

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

defget_db():
    db = database.SessionLocal()
try:
yield db
finally:
        db.close()

defget_current_user(
    token:str= Depends(oauth2_scheme),
    db: Session = Depends(get_db)
)-> models.User:
    username = auth.decode_token(token)
if username isNone:
raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate":"Bearer"},
)
    user = db.query(models.User).filter(models.User.username == username).first()
if user isNone:
raise HTTPException(status_code=404, detail="User not found")
return user

6. routers/auth.py - 认证路由

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from..import schemas, models, auth, dependencies

router = APIRouter(prefix="/auth", tags=["authentication"])

@router.post("/register", response_model=schemas.User)
defregister(user: schemas.UserCreate, db: Session = Depends(dependencies.get_db)):
    db_user = db.query(models.User).filter(models.User.username == user.username).first()
if db_user:
raise HTTPException(status_code=400, detail="Username already registered")
    hashed = auth.get_password_hash(user.password)
    new_user = models.User(username=user.username, hashed_password=hashed)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
return new_user

@router.post("/token", response_model=schemas.Token)
deflogin(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(dependencies.get_db)):
    user = db.query(models.User).filter(models.User.username == form_data.username).first()
ifnot user ornot auth.verify_password(form_data.password, user.hashed_password):
raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token = auth.create_access_token(data={"sub": user.username})
return{"access_token": access_token,"token_type":"bearer"}

7. routers/users.py - 用户信息路由

from fastapi import APIRouter, Depends
from..import schemas, dependencies

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/me", response_model=schemas.User)
defread_users_me(current_user = Depends(dependencies.get_current_user)):
return current_user

8. routers/chat.py - 聊天 WebSocket 路由

from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from typing import Dict
from..auth import decode_token

router = APIRouter(prefix="/chat", tags=["chat"])

classConnectionManager:
def__init__(self):
        self.active_connections: Dict[str, WebSocket]={}# username -> websocket

asyncdefconnect(self, websocket: WebSocket, username:str):
await websocket.accept()
        self.active_connections[username]= websocket

defdisconnect(self, username:str):
        self.active_connections.pop(username,None)

asyncdefsend_personal(self, message:str, username:str):
if username in self.active_connections:
await self.active_connections[username].send_text(message)

asyncdefbroadcast(self, message:str, exclude_username:str=None):
for username, conn in self.active_connections.items():
if username != exclude_username:
await conn.send_text(message)

manager = ConnectionManager()

@router.websocket("/ws")
asyncdefwebsocket_endpoint(websocket: WebSocket):
# 从查询参数获取 token
    token = websocket.query_params.get("token")
ifnot token:
await websocket.close(code=1008)
return

    username = decode_token(token)
ifnot username:
await websocket.close(code=1008)
return

await manager.connect(websocket, username)
try:
whileTrue:
            data =await websocket.receive_text()
# 广播消息给所有其他用户
await manager.broadcast(f"{username}: {data}", exclude_username=username)
except WebSocketDisconnect:
        manager.disconnect(username)
await manager.broadcast(f"User {username} left the chat")

9. main.py - 应用入口

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from.database import engine
from.models import Base
from.routers import auth, users, chat

# 创建数据库表
Base.metadata.create_all(bind=engine)

app = FastAPI(title="Chat App with JWT & WebSocket")

# 配置 CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],# 生产环境应限制
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 包含路由
app.include_router(auth.router)
app.include_router(users.router)
app.include_router(chat.router)

@app.get("/")
defroot():
return{"message":"Welcome to Chat App API"}

10. README.md - 项目说明

# Chat App with FastAPI

一个简单的实时聊天应用，包含用户认证和 WebSocket 通信。

## 功能

- 用户注册
- 用户登录（获取 JWT 令牌）
- 获取当前用户信息
- WebSocket 聊天室（需令牌验证）

## 快速开始

1. 克隆仓库
2. 安装依赖：`pip install -r requirements.txt`
3. 运行服务：`uvicorn chat_app.main:app --reload`
4. 访问 API 文档：http://localhost:8000/docs
5. 使用 WebSocket 客户端连接 `ws://localhost:8000/chat/ws?token=<your_jwt_token>` 进行聊天

## API 端点

- `POST /auth/register` - 注册新用户
- `POST /auth/token` - 登录获取令牌
- `GET /users/me` - 获取当前用户信息（需要认证）
- WebSocket `/chat/ws` - 聊天室连接（需提供 token 查询参数）

运行与测试

1. 启动服务：

   uvicorn chat_app.main:app --reload
   

2. 注册用户：
   使用 /auth/register 端点，POST JSON：

   {"username":"alice","password":"secret"}
   

3. 登录获取令牌：
   POST /auth/token，表单数据：

   username: alice
   password: secret
   

   返回类似：

   {"access_token":"eyJhbGc...","token_type":"bearer"}
   

4. WebSocket 连接：
   使用 WebSocket 客户端（如浏览器 JavaScript、Postman、wscat）连接到：

   ws://localhost:8000/chat/ws?token=eyJhbGc...
   

   连接成功后，可以发送消息，所有其他在线用户将收到广播。

5. 测试受保护路由：
   在请求头中添加 Authorization: Bearer <token> 访问 /users/me。

注意事项

• 生产环境中请勿将密钥硬编码，应使用环境变量。
• 数据库使用 SQLite，仅适用于开发测试。
• WebSocket 连接目前不存储聊天记录，仅实时转发。

通过这个完整项目，你可以看到 FastAPI 的高级特性如何在实际中落地：JWT 认证保护 HTTP 和 WebSocket 端点，SQLAlchemy 处理数据持久化，依赖项简化代码复用，WebSocket 实现实时通信。希望这个示例能帮助你进一步掌握 FastAPI，构建出更强大的应用！

欢迎关注公众号，感谢对文章的点赞分享喜欢，冉成未来会持续更新前后端开发技术、人工智能技术、IT相关的文章及学习经验、知识分享，未来虽然充满着不确定性，但我们可以不断提升自己，不断为未来做准备，让未来更好的自己成就更美好的未来。