FastAPI进阶:从会写接口到生产级架构的完整指南

学会写API接口不难，但离"能上线的API"还有一段距离。假设你写了一个博客系统，基础篇教你的东西能让你完成CRUD，但现实中的问题是：接口没有认证需要OAuth2+JWT、每个接口都写一遍请求日志需要中间件、用户注册完要等邮件发送完才能响应需要后台任务、前端想要实时收到新评论通知需要WebSocket、改完代码不敢上线需要测试。

生产级架构的四大支柱

FastAPI进阶篇的核心，是教你从"会写接口"到"能上线的API"的完整蜕变。

1. OAuth2 + JWT认证体系

基础接口没有认证意味着任何人都可以删除文章。OAuth2密码流的工作原理是：用户拿用户名+密码换Token，然后每次请求都带上Token。

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# Token生成
def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# Token验证（FastAPI自动从请求头提取）
async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    return {"username": username}

# 受保护的路由
@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
    return current_user

2. 中间件：请求日志与统一处理

每个接口都写一遍请求日志不现实。中间件的本质是一个函数，接收request和一个call_next参数，可以在这里统一处理所有请求。

from fastapi import FastAPI, Request
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    
    # 请求处理前
    logger.info(f"请求开始: {request.method} {request.url}")
    
    response = await call_next(request)
    
    # 请求处理后
    process_time = time.time() - start_time
    logger.info(
        f"请求完成: {request.method} {request.url} "
        f"状态码: {response.status_code} 耗时: {process_time:.3f}s"
    )
    
    response.headers["X-Process-Time"] = str(process_time)
    return response

3. BackgroundTasks：异步任务处理

用户注册完要等邮件发送完才能响应？不需要。FastAPI内置的BackgroundTasks是轻量级异步任务工具，响应优先返回，任务在后台默默执行。

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import logging

logger = logging.getLogger(__name__)

app = FastAPI()

class UserRegister(BaseModel):
    username: str
    email: str

def send_welcome_email(email: str):
    """后台任务:发送欢迎邮件"""
    # 实际场景中这里会调用邮件服务
    logger.info(f"发送欢迎邮件到: {email}")
    # 模拟邮件发送延迟
    import time
    time.sleep(2)
    logger.info(f"邮件发送完成: {email}")

@app.post("/register")
async def register(user: UserRegister, background_tasks: BackgroundTasks):
    # 先返回响应
    background_tasks.add_task(send_welcome_email, user.email)
    return {"message": "注册成功", "username": user.username}

4. WebSocket：实时双向通信

HTTP是"你问我答"，WebSocket是"打电话"。建立连接后，双方随时可以说话。适合聊天、实时通知、协作编辑等场景。

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []
    
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
    
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
    
    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            # 广播给所有连接的客户端
            await manager.broadcast(f"客户端 {client_id}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"客户端 {client_id} 断开连接")

Lifespan：应用生命周期管理

应用启动时需要连接数据库、加载模型，关闭时需要释放资源。通过@asynccontextmanager实现生命周期钩子。

from contextlib import asynccontextmanager
from fastapi import FastAPI
import logging

logger = logging.getLogger(__name__)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时执行
    logger.info("应用启动中...")
    # 连接数据库
    # 加载机器学习模型
    # 初始化缓存连接
    yield
    # 关闭时执行
    logger.info("应用关闭中...")
    # 关闭数据库连接
    # 保存模型状态
    # 清理缓存

app = FastAPI(lifespan=lifespan)

@app.get("/")
async def root():
    return {"message": "FastAPI生产级架构示例"}

总结

从基础API到生产级架构，核心是解决四个问题：认证安全、日志追踪、异步任务、实时通信。FastAPI提供了完整的解决方案，配合Uvicorn/Gunicorn部署，即可构建高可用、高性能的Web服务。