系列：Python Web 框架
FastAPI 安装与使用指南 覆盖了路由、参数校验和自动文档等基础功能。本文继续深入，介绍依赖注入、中间件、后台任务、异步数据库整合、错误处理和生产部署——这些是「把 FastAPI 项目真正上线」必须了解的内容。
本文基于 FastAPI >=0.110.0、Pydantic V2（>=2.7.0）、SQLModel >=0.0.14 的当前（2026 年）技术栈，部分示例可直接用于生产环境。
依赖注入进阶
FastAPI 的依赖注入系统是整个框架的核心。掌握好它，代码的可测试性和复用性会有质的提升。
链式依赖
依赖可以层层嵌套，每个环节独立可测：
from fastapi import Depends, HTTPException, Securityfrom fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
security = HTTPBearer()
async def get_token( credentials: HTTPAuthorizationCredentials = Security(security),) -> str: return credentials.credentials
async def get_current_user(token: str = Depends(get_token)) -> dict: user = await decode_jwt(token) # 伪代码：解析 JWT if not user: raise HTTPException(status_code=401, detail="无效的认证令牌") return user
async def require_admin(user: dict = Depends(get_current_user)) -> dict: if user.get("role") != "admin": raise HTTPException(status_code=403, detail="需要管理员权限") return user
@app.get("/admin/dashboard")async def admin_dashboard(user: dict = Depends(require_admin)): return {"message": f"欢迎，管理员 {user['name']}"}这种链式设计的好处：每个 Depends 函数职责单一，可以独立测试，也可以在不同路由中组合复用。
类依赖（Callable Class）
当依赖需要持有状态或配置时，用类比函数更灵活：
class PaginationParams: def __init__(self, max_size: int = 100): self.max_size = max_size
def __call__(self, page: int = 1, size: int = 20): if size > self.max_size: raise HTTPException( status_code=400, detail=f"每页最多 {self.max_size} 条", ) if page < 1: raise HTTPException(status_code=400, detail="page 必须 >= 1") return {"page": page, "size": size, "offset": (page - 1) * size}
@app.get("/items")async def list_items(pagination: dict = Depends(PaginationParams(max_size=50))): # pagination = {"page": 1, "size": 20, "offset": 0} ...也可用于封装数据库连接、RPC 客户端等有状态资源。
请求作用域缓存
FastAPI 在同一请求内只会执行一次依赖（默认 use_cache=True）：
def get_db(): db = SessionLocal() try: yield db finally: db.close()
def get_current_user(db=Depends(get_db), token=Depends(get_token)): ...
# get_db 和 get_token 分别在本次请求中只执行一次@app.get("/profile")def profile(db=Depends(get_db), user=Depends(get_current_user)): ...如果你需要每次调用都执行，设置 Depends(func, use_cache=False)。
测试时覆盖依赖
FastAPI 提供了 dependency_overrides，测试时不需要 mock/patch：
from fastapi.testclient import TestClient
app.dependency_overrides[get_current_user] = lambda: { "id": 1, "name": "测试用户", "role": "admin"}
client = TestClient(app)
def test_admin_dashboard(): response = client.get("/admin/dashboard") assert response.status_code == 200 assert "测试用户" in response.json()["message"]这是 FastAPI 测试哲学的核心：用依赖注入替代 mock，测试代码和业务代码一样干净。
中间件
请求 ID 中间件
每个请求分配唯一 ID，方便日志追踪和问题排查：
import uuidfrom starlette.middleware.base import BaseHTTPMiddleware
class RequestIDMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): request_id = request.headers.get("X-Request-ID", str(uuid.uuid4())) request.state.request_id = request_id response = await call_next(request) response.headers["X-Request-ID"] = request_id return response
app.add_middleware(RequestIDMiddleware)请求计时中间件
import time
class TimingMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): start = time.perf_counter() response = await call_next(request) elapsed = (time.perf_counter() - start) * 1000 response.headers["X-Process-Time-Ms"] = f"{elapsed:.2f}" return response注意：BaseHTTPMiddleware 的性能开销
BaseHTTPMiddleware 基于 StreamingResponse，每次请求会引入额外的数据复制。对于极高吞吐场景，建议用纯 ASGI 中间件：
class ASGITimingMiddleware: def __init__(self, app): self.app = app
async def __call__(self, scope, receive, send): if scope["type"] != "http": return await self.app(scope, receive, send)
start = time.perf_counter()
async def send_wrapper(message): if message["type"] == "http.response.start": elapsed = (time.perf_counter() - start) * 1000 headers = list(message.get("headers", [])) headers.append((b"x-process-time-ms", f"{elapsed:.2f}".encode())) message["headers"] = headers await send(message)
await self.app(scope, receive, send_wrapper)纯 ASGI 中间件零额外开销，适合高并发场景。
Lifespan 事件
FastAPI 已弃用 @app.on_event("startup") 和 @app.on_event("shutdown")，推荐使用 lifespan：
from contextlib import asynccontextmanagerfrom fastapi import FastAPI
@asynccontextmanagerasync def lifespan(app: FastAPI): # 启动：初始化数据库连接池、Redis 客户端、HTTP 客户端等 app.state.db_pool = await create_db_pool() app.state.redis = await create_redis_client() app.state.http_client = httpx.AsyncClient() yield # 关闭：释放资源 await app.state.db_pool.close() await app.state.redis.close() await app.state.http_client.aclose()
app = FastAPI(lifespan=lifespan)从旧事件钩子迁移时，核心变化是把启动和关闭逻辑收敛到同一个上下文管理器：
@app.on_event("startup")async def startup(): app.state.redis = await create_redis_client()
@app.on_event("shutdown")async def shutdown(): await app.state.redis.close()@asynccontextmanagerasync def lifespan(app: FastAPI): app.state.redis = await create_redis_client() yield await app.state.redis.close()
app = FastAPI(lifespan=lifespan)在路由中通过 request.app.state 访问这些资源：
@app.get("/cache/{key}")async def get_cache(key: str, request: Request): value = await request.app.state.redis.get(key) return {"key": key, "value": value}异步数据库整合
FastAPI 的推荐数据库方案是 SQLModel（由 FastAPI 作者开发，融合了 SQLAlchemy 和 Pydantic）：
统一模型定义
SQLModel 的一个模型同时是数据库表定义、请求校验和响应序列化：
from sqlmodel import SQLModel, Field
# 数据库模型class Post(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) title: str = Field(index=True) body: str published: bool = Field(default=False)
# 请求体（只暴露需要客户端填写的字段）class PostCreate(SQLModel): title: str = Field(..., min_length=1, max_length=200) body: str published: bool = False
# 响应体class PostPublic(SQLModel): id: int title: str published: bool异步引擎 + 事务管理
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSessionfrom sqlmodel.ext.asyncio.session import AsyncSessionfrom collections.abc import AsyncGenerator
engine = create_async_engine( "postgresql+asyncpg://user:pass@localhost/dbname", pool_pre_ping=True, pool_recycle=300,)
async def get_session() -> AsyncGenerator[AsyncSession, None]: async with AsyncSession(engine) as session: try: yield session await session.commit() except Exception: await session.rollback() raise这个依赖函数统一管理事务：路由正常返回时自动 commit()，异常时 rollback()。所有写操作只需要 session.add() / session.flush()，不需要手动提交。
分层架构
生产项目建议按领域分层（由 Netflix Dispatch 团队推广的模式）：
src/├── posts/│ ├── router.py # HTTP 层│ ├── service.py # 业务逻辑│ ├── repository.py # 数据访问│ ├── schemas.py # Pydantic 模型│ └── models.py # SQLModel 表定义├── users/│ └── ...├── dependencies.py # 全局依赖├── database.py # 引擎和会话└── main.py # 应用入口各层职责：
| 层 | 返回类型 | 异常 |
|---|---|---|
| Router | Pydantic Schema | 不抛异常，调用 Service |
| Service | Pydantic Schema | 抛 DomainException |
| Repository | Model 或 None | 不抛异常，只做数据访问 |
示例：
class PostRepository: async def get_by_id(self, session: AsyncSession, post_id: int) -> Post | None: return await session.get(Post, post_id)
async def list_all(self, session: AsyncSession, offset: int, limit: int) -> list[Post]: result = await session.execute( select(Post).offset(offset).limit(limit).order_by(Post.id.desc()) ) return list(result.scalars().all())
# service.pyclass PostService: def __init__(self, repo: PostRepository = Depends()): self.repo = repo
async def get_post(self, session: AsyncSession, post_id: int) -> PostPublic: post = await self.repo.get_by_id(session, post_id) if post is None: raise NotFoundError("文章", post_id) return PostPublic.model_validate(post)
# router.pyrouter = APIRouter(prefix="/posts", tags=["文章"])
@router.get("/{post_id}", response_model=PostPublic)async def get_post( post_id: int, session: AsyncSession = Depends(get_session), service: PostService = Depends(),): return await service.get_post(session, post_id)这里 PostPublic.model_validate(post) 是 Pydantic V2 的新写法，替代旧的 .from_orm()。
后台任务与任务队列
BackgroundTasks（轻量级）
适合发邮件、写日志等不关心结果的异步操作：
from fastapi import BackgroundTasks
def send_verification_email(email: str, code: str): # 同步函数会在线程池中执行 ...
@app.post("/register")async def register(email: str, background_tasks: BackgroundTasks): background_tasks.add_task(send_verification_email, email, "123456") return {"message": "注册成功，验证邮件已发送"}局限性：没有重试机制、不持久化、服务重启即丢失。
任务队列（生产级）
对于需要重试、调度、监控的任务，使用 Celery / ARQ / Dramatiq：
# 使用 ARQ（异步优先的任务队列，和 FastAPI 配合更好）from arq import create_poolfrom arq.jobs import Job
async def send_report(user_id: int): # 耗时操作 ...
@app.post("/reports/generate")async def generate_report(user_id: int, request: Request): redis = request.app.state.redis # ARQ 也基于 Redis job = await redis.enqueue_job("send_report", user_id) return {"job_id": job.job_id, "status": "queued"}| 特性 | BackgroundTasks | ARQ / Celery |
|---|---|---|
| 重试 | 不支持 | 支持 |
| 持久化 | 不支持 | 支持 |
| 进度追踪 | 不支持 | 支持 |
| 定时任务 | 不支持 | 支持 |
| 适用场景 | 发邮件、记日志 | 报表生成、数据处理、定时任务 |
结构化错误处理
定义统一的异常体系，让 API 错误响应一致：
from fastapi import Requestfrom fastapi.responses import JSONResponse
class AppError(Exception): def __init__(self, message: str, code: str, status_code: int = 400): self.message = message self.code = code self.status_code = status_code
class NotFoundError(AppError): def __init__(self, resource: str, resource_id: int): super().__init__( message=f"{resource}（id={resource_id}）未找到", code="RESOURCE_NOT_FOUND", status_code=404, )
class ConflictError(AppError): def __init__(self, message: str): super().__init__(message=message, code="RESOURCE_CONFLICT", status_code=409)
@app.exception_handler(AppError)async def app_error_handler(request: Request, exc: AppError): return JSONResponse( status_code=exc.status_code, content={ "error": { "code": exc.code, "message": exc.message, "request_id": getattr(request.state, "request_id", None), } }, )
@app.exception_handler(ValidationError)async def validation_error_handler(request: Request, exc: ValidationError): return JSONResponse( status_code=422, content={ "error": { "code": "VALIDATION_ERROR", "message": "请求参数校验失败", "details": exc.errors(), "request_id": getattr(request.state, "request_id", None), } }, )现在业务代码中直接 raise NotFoundError("文章", 42)，客户端会收到结构一致的错误响应，包含 request_id 方便定位问题。
WebSocket 支持
FastAPI 原生支持 WebSocket，适合实时推送场景：
from fastapi import WebSocket, WebSocketDisconnect
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/chat")async def chat(websocket: WebSocket): await manager.connect(websocket) try: while True: data = await websocket.receive_text() await manager.broadcast(f"收到消息: {data}") except WebSocketDisconnect: manager.disconnect(websocket)性能优化清单
根据 2026 年社区实践，性能优化的优先顺序：
- 用纯 ASGI 中间件替代
BaseHTTPMiddleware：减少每请求的 StreamingResponse 开销 - 使用 Pydantic V2：底层 Rust 实现，校验速度比 V1 快 5-50 倍
- 数据库连接池调优：
pool_size和max_overflow按实际负载设置 - 为高频读接口加 Redis 缓存：减少数据库压力
- 正确区分
async def和def：- I/O 密集（数据库、HTTP 调用）→
async def+await - 同步 SDK 无法 await →
def（FastAPI 在线程池中执行），或asyncio.to_thread() - CPU 密集 → 不要在线程池跑，用 Celery/ARQ 推到独立 Worker
- I/O 密集（数据库、HTTP 调用）→
import asyncio
# 在线程池中执行同步的 CPU 密集型操作result = await asyncio.to_thread(process_image_sync, image_bytes)生产部署检查清单
- Lifespan 管理资源：数据库连接池、Redis、HTTP 客户端都在 lifespan 中创建和销毁
- 结构化日志：每个请求携带
request_id，用structlog或loguru输出 JSON 格式日志 - UV / Ruff：用
ruff做 linting 和格式化，uv管理依赖 - 异步测试：用
httpx.AsyncClient+ASGITransport写异步测试，用dependency_overrides隔离依赖 - 健康检查端点：
/health返回数据库和 Redis 连通状态，供负载均衡探测 - CORS 限制：生产环境
allow_origins指定具体域名，不要用["*"] - 速率限制：用
slowapi或 Redis + 自定义中间件做 API 限流 - 静态资源：交给 Nginx 或 CDN，不要用 FastAPI 直接 serve
# 生产环境启动命令uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 --log-level info也可以配合 Gunicorn 管理进程：
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000常见问题
Pydantic V2 迁移
V2 有几个不兼容变更需要注意：
# 旧 .from_orm() → 新 .model_validate()user = UserSchema.model_validate(orm_instance)
# 旧 @validator → 新 @field_validatorfrom pydantic import field_validator
@field_validator("password")@classmethoddef check_strength(cls, v): ...
# 旧 dict() → 新 model_dump()data = user.model_dump()async def 路由中调用了同步 ORM
如果你的 ORM 不支持异步，有两个选择：
- 将路由改为
def（FastAPI 会自动在线程池中运行） - 使用
asyncio.to_thread()包装同步调用：
@app.get("/users/{user_id}")async def get_user(user_id: int): user = await asyncio.to_thread(sync_get_user_from_db, user_id) return user数据库会话在响应阶段已被关闭
不要在响应模型的 computed_field 或 field_serializer 中访问数据库——此时会话可能已关闭。确保所有数据查询在 Service 层完成，Schema 只做纯数据转换。
项目结构混乱
早期不要过度设计。随着代码量增长，按以下节奏演进：
- 单文件 (< 10 路由) →
main.py就够了 - 10-30 路由 → 拆
routers/+schemas.py - 30+ 路由 → 按领域拆包 + Service/Repository 分层
总结
FastAPI 的进阶用法围绕几个核心模式展开：
- 依赖注入：链式依赖 + 类依赖 +
dependency_overrides测试，是 FastAPI 最重要的设计模式 - 中间件：请求 ID、计时用纯 ASGI 中间件，避免
BaseHTTPMiddleware的性能开销 - Lifespan：替代已弃用的
on_event，统一管理资源生命周期 - SQLModel：一个模型同时做 ORM 表定义、请求校验和响应序列化，减少重复代码
- 分层架构：Router → Service → Repository，各层职责清晰
- 结构化异常：自定义异常 + 全局异常处理器，让错误响应一致、可追踪
掌握这些模式后，FastAPI 项目可以从原型平滑演进到生产级服务。需要补齐登录闭环时，可以继续看 FastAPI 项目实战：用户登录与 JWT 鉴权；如果项目已经开始变大，Python Web 项目结构怎么设计 会更适合接着读。更多细节参考 FastAPI 官方文档 和 FastAPI Best Practices。