FastAPI 完全指南

官方文档：https://fastapi.tiangolo.com/zh/
最后更新：2026-03-05

1. 基础概念

FastAPI 是什么

FastAPI 是一个用于构建 API 的现代、高性能 Python Web 框架，基于标准 Python 类型提示。主要特点：

• 基于 ASGI（Starlette），支持异步
• 自动生成交互式 API 文档（Swagger UI / ReDoc）
• 利用 Pydantic 进行数据验证和序列化
• 性能接近 NodeJS 和 Go（基于 Starlette 和 Pydantic）

与 Flask / Django 的对比

ASGI

ASGI（Asynchronous Server Gateway Interface）是 WSGI 的异步升级版。FastAPI 基于 Starlette，Starlette 实现了 ASGI 接口。运行 FastAPI 需要 ASGI 服务器，如 uvicorn 或 hypercorn。

自动文档

FastAPI 自动根据路由和 Pydantic 模型生成两套文档：

• Swagger UI：访问 /docs，支持在线测试接口
• ReDoc：访问 /redoc，适合阅读文档
• OpenAPI JSON：访问 /openapi.json，原始 schema

2. 快速开始

安装

# 安装 FastAPI 和 ASGI 服务器
pip install fastapi uvicorn[standard]

# 可选：安装 python-multipart（文件上传需要）
pip install python-multipart

# 可选：安装 python-jose（JWT 认证需要）
pip install python-jose[cryptography] passlib[bcrypt]

第一个应用

from fastapi import FastAPI

app = FastAPI(
    title="示例 API",          # 文档标题
    description="这是一个示例",  # 文档描述
    version="1.0.0",           # API 版本
)

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

运行（uvicorn 参数表）

uvicorn main:app --reload

3. 路径操作

路径参数

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(
        ...,               # 必填（... 表示必填）
        title="商品ID",
        description="商品的唯一标识符",
        ge=1,              # 大于等于 1
        le=1000,           # 小于等于 1000
    )
):
    return {"item_id": item_id}

Path() 参数表：

查询参数

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(
    q: str = Query(None, min_length=3, max_length=50),
    skip: int = Query(0, ge=0),
    limit: int = Query(10, le=100),
    tags: list[str] = Query(default=[]),  # 支持多值：?tags=a&tags=b
):
    return {"q": q, "skip": skip, "limit": limit, "tags": tags}

Query() 参数表：

请求体（Pydantic BaseModel）

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

@app.post("/items/")
async def create_item(item: Item):
    # item 是经过验证的 Item 实例
    item_dict = item.model_dump()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

4. Pydantic 模型

Field() 参数表

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(
        ...,
        title="商品名称",
        description="商品的显示名称",
        min_length=1,
        max_length=100,
        examples=["笔记本电脑"],
    )
    price: float = Field(..., gt=0, description="价格，必须大于0")
    quantity: int = Field(default=0, ge=0)

类型验证与可选字段

from typing import Optional
from pydantic import BaseModel

class UserProfile(BaseModel):
    username: str                           # 必填
    email: str                              # 必填
    age: Optional[int] = None               # 可选，默认 None（Python 3.9 以下写法）
    bio: str | None = None                  # 可选，默认 None（Python 3.10+ 写法）
    tags: list[str] = []                    # 可选，默认空列表
    metadata: dict[str, str] = {}           # 可选，默认空字典

嵌套模型

from pydantic import BaseModel

class Address(BaseModel):
    street: str
    city: str
    country: str = "China"

class User(BaseModel):
    name: str
    address: Address                        # 嵌套模型
    addresses: list[Address] = []           # 嵌套模型列表

validator / @field_validator（Pydantic v2）

from pydantic import BaseModel, field_validator, model_validator

class UserCreate(BaseModel):
    username: str
    password: str
    confirm_password: str
    email: str

    # 单字段验证（Pydantic v2 写法）
    @field_validator("username")
    @classmethod
    def username_must_be_alphanumeric(cls, v: str) -> str:
        if not v.isalnum():
            raise ValueError("用户名只能包含字母和数字")
        return v.lower()  # 可以对值进行转换

    @field_validator("email")
    @classmethod
    def email_must_contain_at(cls, v: str) -> str:
        if "@" not in v:
            raise ValueError("邮箱格式不正确")
        return v

    # 多字段交叉验证（Pydantic v2 写法）
    @model_validator(mode="after")
    def check_passwords_match(self) -> "UserCreate":
        if self.password != self.confirm_password:
            raise ValueError("两次密码不一致")
        return self

注意：Pydantic v1 使用 @validator，v2 使用 @field_validator，两者语法不同，详见第 17 节。

5. 响应模型

response_model

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserCreate(BaseModel):
    username: str
    password: str          # 输入模型包含密码

class UserPublic(BaseModel):
    id: int
    username: str          # 输出模型不暴露密码

@app.post("/users/", response_model=UserPublic)
async def create_user(user: UserCreate):
    # 返回的数据会按 UserPublic 过滤，password 不会出现在响应中
    return {"id": 1, "username": user.username, "password": user.password}

response_model_exclude_unset

@app.patch("/users/{user_id}", response_model=UserPublic, response_model_exclude_unset=True)
async def update_user(user_id: int, user: UserCreate):
    # 只返回实际设置的字段，未设置的字段不出现在响应中
    return user

状态码

from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
    return {"name": name}

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    return None  # 204 不返回响应体

常用状态码常量（fastapi.status 模块）：

6. 依赖注入

Depends() 基础用法

from fastapi import FastAPI, Depends

app = FastAPI()

# 定义依赖函数
async def get_current_user(token: str = Query(...)):
    # 从 token 中解析用户
    user = verify_token(token)
    return user

@app.get("/users/me")
async def read_users_me(current_user = Depends(get_current_user)):
    return current_user

类依赖

class CommonQueryParams:
    def __init__(self, skip: int = 0, limit: int = 10, q: str | None = None):
        self.skip = skip
        self.limit = limit
        self.q = q

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    # 可简写为 Depends()，FastAPI 自动推断
    return {"skip": commons.skip, "limit": commons.limit, "q": commons.q}

多层依赖

async def verify_token(token: str = Header(...)):
    if token != "secret":
        raise HTTPException(status_code=401, detail="无效 token")
    return token

async def verify_key(api_key: str = Header(...)):
    if api_key != "my-api-key":
        raise HTTPException(status_code=403, detail="无效 API Key")
    return api_key

@app.get("/protected/")
async def protected_route(
    token = Depends(verify_token),
    key = Depends(verify_key),
):
    return {"token": token, "key": key}

数据库会话依赖（与 SQLAlchemy 集成）

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session  # yield 使依赖支持上下文管理器

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    # 请求结束后 session 自动关闭
    user = await db.get(User, user_id)
    return user

详见 SQLAlchemy完全指南。

7. 路由

APIRouter

# routers/items.py
from fastapi import APIRouter

router = APIRouter(
    prefix="/items",        # 所有路由加前缀 /items
    tags=["items"],         # 文档中的分组标签
    dependencies=[Depends(verify_token)],  # 该路由下所有接口共用的依赖
    responses={404: {"description": "Not found"}},  # 通用响应说明
)

@router.get("/")
async def list_items():
    return []

@router.get("/{item_id}")
async def get_item(item_id: int):
    return {"item_id": item_id}

APIRouter() 参数表：

路由注册

# main.py
from fastapi import FastAPI
from routers import items, users

app = FastAPI()

app.include_router(items.router)
app.include_router(users.router, prefix="/api/v1")  # 可覆盖前缀

8. 中间件

@app.middleware("http")

import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)  # 调用下一个处理器
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

CORSMiddleware

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

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

CORSMiddleware 参数表：

自定义中间件（基于类）

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        token = request.headers.get("Authorization")
        if not token:
            from starlette.responses import JSONResponse
            return JSONResponse({"detail": "未授权"}, status_code=401)
        response = await call_next(request)
        return response

app.add_middleware(AuthMiddleware)

9. 认证

OAuth2PasswordBearer

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

app = FastAPI()

# tokenUrl 是获取 token 的接口路径
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

@app.post("/auth/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # form_data.username, form_data.password
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="用户名或密码错误")
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

JWT Token（python-jose）

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

SECRET_KEY = "your-secret-key"  # 生产环境使用环境变量
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def decode_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="Token 无效或已过期")

依赖注入保护路由

from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer

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

async def get_current_user(token: str = Depends(oauth2_scheme)):
    payload = decode_token(token)
    username = payload.get("sub")
    if not username:
        raise HTTPException(status_code=401, detail="无效凭证")
    user = await get_user_by_username(username)
    if not user:
        raise HTTPException(status_code=401, detail="用户不存在")
    return user

async def get_current_active_user(current_user = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="用户已禁用")
    return current_user

@app.get("/users/me")
async def read_users_me(current_user = Depends(get_current_active_user)):
    return current_user

10. 文件上传

单文件上传

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()          # 读取文件内容（bytes）
    return {
        "filename": file.filename,        # 原始文件名
        "content_type": file.content_type, # MIME 类型
        "size": len(contents),
    }

UploadFile 属性和方法

多文件上传

@app.post("/upload/multiple/")
async def upload_multiple_files(files: list[UploadFile] = File(...)):
    results = []
    for file in files:
        contents = await file.read()
        results.append({"filename": file.filename, "size": len(contents)})
    return results

保存文件到磁盘

import aiofiles

@app.post("/upload/save/")
async def upload_and_save(file: UploadFile = File(...)):
    save_path = f"/tmp/{file.filename}"
    async with aiofiles.open(save_path, "wb") as f:
        while chunk := await file.read(1024 * 1024):  # 每次读取 1MB
            await f.write(chunk)
    return {"saved_to": save_path}

11. WebSocket

WebSocket 连接

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()             # 接受连接
    try:
        while True:
            data = await websocket.receive_text()   # 接收文本
            await websocket.send_text(f"收到: {data}")  # 发送文本
    except Exception:
        await websocket.close()

广播示例（多客户端）

from fastapi import FastAPI, WebSocket
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 Exception:
        manager.disconnect(websocket)
        await manager.broadcast(f"客户端 {client_id} 已离线")

12. 后台任务

BackgroundTasks.add_task()

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def send_email_notification(email: str, message: str):
    # 模拟耗时操作，在后台线程执行（同步函数）
    import time
    time.sleep(5)
    print(f"发送邮件到 {email}: {message}")

async def log_operation(operation: str):
    # 异步后台任务
    await some_db_operation(operation)

@app.post("/send-notification/")
async def send_notification(
    email: str,
    background_tasks: BackgroundTasks,
):
    # add_task 的参数：func, *args, **kwargs
    background_tasks.add_task(send_email_notification, email, "操作成功")
    background_tasks.add_task(log_operation, "send_notification")
    return {"message": "通知已加入队列"}

注意：BackgroundTasks 在响应返回后在同一进程中执行，适合轻量任务。重量级任务请使用 Celery 或 ARQ。

13. 异常处理

HTTPException

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(
            status_code=404,
            detail="商品不存在",         # 可以是字符串或字典
            headers={"X-Error": "true"},  # 可选：附加响应头
        )
    return items_db[item_id]

自定义异常处理器

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

class BusinessException(Exception):
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message

@app.exception_handler(BusinessException)
async def business_exception_handler(request: Request, exc: BusinessException):
    return JSONResponse(
        status_code=400,
        content={"code": exc.code, "message": exc.message},
    )

@app.exception_handler(404)
async def not_found_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=404,
        content={"message": "资源不存在"},
    )

覆盖默认验证错误处理

from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "body": exc.body},
    )

14. 测试

TestClient（httpx）

# test_main.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_create_item():
    response = client.post(
        "/items/",
        json={"name": "笔记本", "price": 99.9},
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "笔记本"

pytest fixture（依赖注入覆盖）

import pytest
from fastapi.testclient import TestClient
from main import app
from dependencies import get_db

# 测试用数据库
def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

@pytest.fixture
def client():
    # 覆盖依赖项
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

def test_create_user(client):
    response = client.post("/users/", json={"username": "test", "password": "secret"})
    assert response.status_code == 201

异步测试

import pytest
import httpx
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_async_endpoint():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/")
    assert response.status_code == 200

15. 部署

uvicorn + gunicorn

# 安装
pip install gunicorn uvicorn[standard]

# 运行（gunicorn 管理进程，uvicorn 作为 worker）
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

Docker

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# 生产环境运行
CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]

环境变量配置（pydantic-settings）

# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str = "sqlite:///./test.db"
    secret_key: str = "dev-secret-key"
    debug: bool = False
    allowed_origins: list[str] = ["http://localhost:3000"]

    class Config:
        env_file = ".env"           # 从 .env 文件读取
        env_file_encoding = "utf-8"

settings = Settings()

# 使用
@app.get("/info")
async def info():
    return {"debug": settings.debug}

.env 文件：

DATABASE_URL=postgresql+asyncpg://user:password@localhost/mydb
SECRET_KEY=production-secret-key-change-this
DEBUG=false

16. 最佳实践

项目目录结构

myproject/
├── main.py                 # FastAPI 应用入口
├── config.py               # 配置管理（pydantic-settings）
├── database.py             # 数据库连接和 session
├── models/                 # SQLAlchemy ORM 模型
│   ├── __init__.py
│   ├── user.py
│   └── item.py
├── schemas/                # Pydantic 输入输出模型
│   ├── __init__.py
│   ├── user.py
│   └── item.py
├── routers/                # APIRouter 路由模块
│   ├── __init__.py
│   ├── users.py
│   └── items.py
├── dependencies/           # 可复用依赖项
│   ├── __init__.py
│   └── auth.py
├── services/               # 业务逻辑层（可选）
│   ├── __init__.py
│   └── user_service.py
└── tests/
    ├── conftest.py
    └── test_users.py

异步数据库（与 SQLAlchemy 集成）

详见 SQLAlchemy完全指南。

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,          # 连接池大小
    max_overflow=20,       # 超出 pool_size 时最多再创建的连接数
    echo=False,            # 生产环境关闭 SQL 日志
)

AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

lifespan（应用生命周期，推荐写法）

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 应用启动时执行
    await create_db_tables()
    print("应用已启动")
    yield
    # 应用关闭时执行
    await engine.dispose()
    print("应用已关闭")

app = FastAPI(lifespan=lifespan)

注意：lifespan 是推荐写法，替代已废弃的 @app.on_event("startup") 和 @app.on_event("shutdown")。

17. 常见陷阱

同步函数 vs async

# 陷阱：在 async 路由中调用同步阻塞函数会阻塞事件循环
@app.get("/bad/")
async def bad_endpoint():
    import time
    time.sleep(5)  # 错误！阻塞整个事件循环
    return {}

# 正确做法 1：使用 asyncio.sleep
import asyncio
@app.get("/good-sleep/")
async def good_sleep():
    await asyncio.sleep(5)  # 正确，不阻塞事件循环
    return {}

# 正确做法 2：阻塞操作放在线程池
import asyncio
@app.get("/good-blocking/")
async def good_blocking():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, blocking_function)
    return {"result": result}

# 正确做法 3：路由函数定义为普通函数（FastAPI 自动放入线程池）
@app.get("/sync-route/")
def sync_route():
    import time
    time.sleep(5)  # 可以，FastAPI 会在线程池中运行
    return {}

Pydantic v1 vs v2 区别

FastAPI 0.100.0+ 默认使用 Pydantic v2，但保留了 v1 兼容层。

CORS 常见错误

# 错误：allow_credentials=True 时不能使用通配符 origins
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],          # 错误！
    allow_credentials=True,
)

# 正确：明确指定来源
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://frontend.example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

路由顺序问题

# 陷阱：固定路径必须在参数路径之前定义
@app.get("/items/search")   # 如果放在 /{item_id} 后面，会被 /{item_id} 匹配到
async def search_items(): ...

@app.get("/items/{item_id}")
async def get_item(item_id: str): ...

Response 对象泄漏

# 陷阱：直接返回 ORM 对象（lazy 属性在 session 关闭后访问失败）
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await db.get(User, user_id)
    return user  # 如果 User 有懒加载关系，此处可能报错

# 正确：使用 response_model 或先序列化
@app.get("/users/{user_id}", response_model=UserPublic)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await db.get(User, user_id)
    return UserPublic.model_validate(user)  # 在 session 关闭前完成序列化

最佳实践

依赖注入用 Annotated 类型封装：将 Depends(get_db) 包在 Annotated 里，路由签名更简洁，且可全局复用：

from typing import Annotated
DbDep = Annotated[AsyncSession, Depends(get_db)]

@router.get("/users")
async def list_users(db: DbDep):
    ...

用 APIRouter 按业务模块拆分路由：不要把所有路由都写在 main.py，每个业务域一个 router 文件，在 main.py 用 app.include_router 聚合，prefix 和 tags 在 include 时统一设置。

response_model 始终显式声明：FastAPI 依赖 response_model 过滤和验证响应，未声明时直接序列化 ORM 对象可能暴露敏感字段（如密码哈希）。

背景任务用 BackgroundTasks，长任务用 Celery：BackgroundTasks 在响应发送后同步运行，适合发邮件等快速任务；CPU 密集或需要重试的任务放 Celery worker，避免阻塞 ASGI worker。

中间件顺序影响请求链：app.add_middleware 后添加的中间件先执行（洋葱模型），CORS 中间件必须最外层（最后 add），否则预检请求会被其他中间件拦截返回非 200。

常见陷阱

陷阱：在路由函数中直接修改 Pydantic 模型实例

现象： 修改从请求体解析的 schema 对象后，后续验证或序列化行为异常。
原因： Pydantic v2 默认 model 是不可变的，直接修改字段会静默失败或抛出验证错误。
解决： 用 model.model_copy(update={...}) 创建修改后的新实例。

陷阱：async def 路由中调用同步阻塞函数

现象： 高并发时 FastAPI 响应明显变慢，/metrics 显示 active requests 堆积。
原因： async def 路由运行在事件循环中，同步阻塞调用（time.sleep、同步 DB 查询）阻塞整个循环。
解决： 同步阻塞代码用 run_in_executor 放线程池，或改用异步库（httpx、asyncpg）。

陷阱：生命周期事件用 @app.on_event 已废弃

现象： 代码中使用 @app.on_event("startup") 后，新版 FastAPI 打印 DeprecationWarning。
原因： FastAPI 0.93+ 推荐用 lifespan context manager 替代 on_event，on_event 将在未来版本移除。
解决： 改用 @asynccontextmanager + lifespan 参数：

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    yield
    await close_db()

app = FastAPI(lifespan=lifespan)

参见

• SQLAlchemy完全指南
• Pydantic完全指南
• asyncio异步编程完全指南
• pytest完全指南
• JWT完全指南