FastAPI + Pydantic v2 严格校验请求与响应（含自定义错误格式）

FastAPI 的核心卖点是"用 Python 类型注解定义 API schema，自动校验 +
生成 OpenAPI 文档"。Pydantic v2 是其背后的校验引擎，比 v1 快 5-50 倍。

1. 最小例子

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, Field

app = FastAPI()

class CreateUser(BaseModel):
    email: EmailStr
    nickname: str = Field(min_length=2, max_length=30)
    age: int = Field(ge=0, le=150)

class UserOut(BaseModel):
    id: int
    email: EmailStr
    nickname: str

@app.post('/users', response_model=UserOut, status_code=201)
def create_user(payload: CreateUser) -> UserOut:
    # ... 写入 DB ...
    return UserOut(id=42, email=payload.email, nickname=payload.nickname)

发请求时任何字段不合法都返回 422 + 详细错误。打开 /docs 看自动文档。

2. 自定义 validator

from pydantic import field_validator

class CreateUser(BaseModel):
    nickname: str

    @field_validator('nickname')
    @classmethod
    def no_whitespace(cls, v: str) -> str:
        if v != v.strip() or '  ' in v:
            raise ValueError('昵称不能含首尾或连续空格')
        return v

V2 必须加 @classmethod。

3. 计算字段（动态）

from pydantic import computed_field

class UserOut(BaseModel):
    nickname: str
    username: str

    @computed_field
    @property
    def display_name(self) -> str:
        return f'{self.nickname}@{self.username}'

4. 加载 / 序列化别名

class UserIn(BaseModel):
    email_address: str = Field(alias='email')
    # 接收的 JSON 里是 "email"，Python 属性是 email_address

5. 严格模式：拒绝多余字段

V2 默认允许额外字段被忽略。生产里建议严格：

class CreateUser(BaseModel):
    model_config = {'extra': 'forbid'}
    email: EmailStr
    # 客户端发 {"email": ..., "foo": ...} 会 422

6. 统一错误格式

FastAPI 默认 422 响应是 Pydantic 原始结构。给前端友好点：

from fastapi import Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    errors = [
        {
            'field': '.'.join(str(x) for x in e['loc']),
            'message': e['msg'],
            'type': e['type'],
        } for e in exc.errors()
    ]
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={'detail': '请求数据校验失败', 'errors': errors},
    )

7. 路径 / 查询 / Body / Header / Cookie

from fastapi import Path, Query, Header, Cookie

@app.get('/items/{item_id}')
def get_item(
    item_id: int = Path(ge=1),
    fields: list[str] | None = Query(None, max_length=10),
    user_agent: str | None = Header(None),
    session: str | None = Cookie(None),
):
    ...

8. Depends 注入

from fastapi import Depends

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

@app.get('/users/{uid}')
def read_user(uid: int, db = Depends(get_db)):
    return db.query(User).get(uid)

Depends 是 FastAPI 的 DI 系统；可以套娃（依赖里又用 Depends）。

9. 鉴权依赖

from fastapi import HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2 = OAuth2PasswordBearer(tokenUrl='token')

def current_user(token: str = Depends(oauth2)):
    user = decode_jwt(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            headers={'WWW-Authenticate': 'Bearer'})
    return user

@app.get('/me')
def me(user = Depends(current_user)):
    return user

10. 后台任务（轻量）

from fastapi import BackgroundTasks

def send_welcome_email(email: str):
    # ... 同步发邮件 ...
    ...

@app.post('/users')
def create_user(payload: CreateUser, tasks: BackgroundTasks):
    user = save(payload)
    tasks.add_task(send_welcome_email, user.email)
    return user

请求立即返回，邮件在响应发出后异步执行。注意：失败没有重试。
要可靠就上 Celery / RQ / Arq。

11. CORS

from fastapi.middleware.cors import CORSMiddleware

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

12. 运行

uv add fastapi 'uvicorn[standard]'
uv run uvicorn main:app --reload  # 开发
uv run uvicorn main:app --workers 4 --host 0.0.0.0 --port 8000  # 生产
# 生产更稳的是用 gunicorn 启 uvicorn worker：
uv run gunicorn -k uvicorn.workers.UvicornWorker -w 4 main:app

踩过的坑

Pydantic v1 / v2 在同一项目混用：v1 model 的 .dict()、.json()
在 v2 是 .model_dump() 和 .model_dump_json()。代码改名后老的方法
调用会静默返回不正确的格式。
用 dataclass 替代 BaseModel：dataclass 在 FastAPI 路径参数处不会被校验，
只在 response_model 处校验。混着用很容易出问题。
response_model 会过滤响应字段（不在 model 里的字段被丢掉），
这是 feature 不是 bug。想全输出就别设 response_model 或用 dict。
BackgroundTasks 是同一进程里的协程，长任务会撑住 worker；
超过 30 秒的任务就该上 Celery。