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

class User(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

class UserBase(BaseModel):
username: str

class UserCreate(UserBase):
password: str

class User(UserBase):
id: int
class Config:
orm_mode = True

class Token(BaseModel):
access_token: str
token_type: str

class TokenData(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")

def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
return pwd_context.hash(password)

def create_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

def decode_token(token: str):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
return None
return username
except JWTError:
return None

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")

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

def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
) –> models.User:
username = auth.decode_token(token)
if username is None:
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 is None:
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)
def register(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)
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(dependencies.get_db)):
user = db.query(models.User).filter(models.User.username == form_data.username).first()
if not user or not 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)
def read_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"])

class ConnectionManager:
def __init__(self):
self.active_connections: Dict[str, WebSocket] = {} # username -> websocket

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

def disconnect(self, username: str):
self.active_connections.pop(username, None)

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

async def broadcast(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")
async def websocket_endpoint(websocket: WebSocket):
# 从查询参数获取 token
token = websocket.query_params.get("token")
if not token:
await websocket.close(code=1008)
return

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

await manager.connect(websocket, username)
try:
while True:
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("/")
def root():
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 查询参数）

运行与测试

注意事项

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

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