高级 Pydantic 与 pydantic-settings

第 13 章 · 高级 Pydantic 与 pydantic-settings

本章目标：掌握 Pydantic v2 model_config（ConfigDict）定制序列化与校验行为；使用 computed_field 派生只读字段；用 TypeAdapter 校验非模型结构（列表、联合类型）；通过 pydantic-settings 的 BaseSettings 统一管理 svc-demo 环境变量；与 FastAPI Depends 注入配置；建立配置分层与 .env 安全规范。

学时建议：4～5 小时（含 1.5 小时重构 svc-demo 配置）

前置：本模块 ch03 Pydantic 基础、ch04 响应模型；已完成 ch12 svc-demo 骨架者可边学边改 core/config.py。

13.1 为什么需要「高级」Pydantic

┌─────────────────────────────────────────────────┐
│  .env / 环境变量  ──►  BaseSettings (ch13)       │
│         │                                        │
│         ▼                                        │
│  FastAPI Depends(get_settings)                   │
│         │                                        │
│         ▼                                        │
│  API Schema ◄── model_config / computed_field    │
│         │                                        │
│         ▼                                        │
│  TypeAdapter ◄── 裸 list/dict、第三方 JSON       │
└─────────────────────────────────────────────────┘

示例项目：svc-demo；禁止将真实密钥写入教程或 Git。

13.2 model_config 与 ConfigDict

Pydantic v2 用 model_config = ConfigDict(...) 替代 v1 的 class Config。

常用配置项

svc_demo/schemas/product.py：

from decimal import Decimal
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field


class ProductBase(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,
        extra="forbid",
    )

    name: str = Field(..., min_length=1, max_length=200)
    slug: str = Field(..., pattern=r"^[a-z0-9-]+$")
    price: Decimal = Field(..., gt=0, decimal_places=2)
    stock: int = Field(0, ge=0)
    description: str | None = None


class ProductCreate(ProductBase):
    is_published: bool = False


class ProductOut(ProductBase):
    model_config = ConfigDict(from_attributes=True)

    id: int
    is_published: bool
    created_at: datetime
    updated_at: datetime

序列化定制

from pydantic import ConfigDict


class TokenOut(BaseModel):
    model_config = ConfigDict(
        # 响应 JSON 时隐藏 refresh_token（若走不同端点可拆分模型）
        json_schema_extra={
            "example": {
                "access_token": "eyJ...",
                "token_type": "Bearer",
                "expires_in": 900,
            }
        }
    )

    access_token: str
    token_type: str = "Bearer"
    expires_in: int

使用 model_dump(mode="json") 输出 API 时，Decimal 自动转为可 JSON 序列化类型（配合 FastAPI jsonable_encoder）。

13.3 computed_field 计算字段

computed_field 像普通字段出现在 OpenAPI，但不参与反序列化入参。

from pydantic import BaseModel, computed_field, ConfigDict
from decimal import Decimal


class ProductOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    name: str
    price: Decimal
    stock: int
    is_published: bool

    @computed_field
    @property
    def display_price(self) -> str:
        return f"¥{self.price:.2f}"

    @computed_field
    @property
    def in_stock(self) -> bool:
        return self.stock > 0

    @computed_field
    @property
    def status_label(self) -> str:
        if not self.is_published:
            return "未上架"
        return "在售" if self.in_stock else "缺货"

注意：computed_field 依赖的字段必须先定义；循环依赖会报错。

带参数的「伪计算」（不推荐）

计算字段不能带参数；复杂逻辑放 service 层，Schema 只暴露结果字段。

13.4 字段别名与 populate_by_name

对接 user-demo 前端 camelCase：

from pydantic import BaseModel, Field, ConfigDict


class ProductQuery(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    page: int = Field(1, ge=1, alias="page")
    per_page: int = Field(10, ge=1, le=50, alias="perPage")
    q: str | None = Field(None, alias="search")

FastAPI 查询参数：

@router.get("")
async def list_products(query: ProductQuery = Depends()):
    ...

Query 传 perPage=20 或 per_page=20 均可（需 populate_by_name=True）。

13.5 TypeAdapter：校验任意类型

当目标不是单一 BaseModel，而是 list、dict、Union 时，用 TypeAdapter。

from pydantic import TypeAdapter
from typing import Annotated
from pydantic import Field

PositiveIntList = Annotated[list[int], Field(min_length=1, max_length=100)]
adapter_ids = TypeAdapter(PositiveIntList)

raw = ["1", "2", "3"]
ids = adapter_ids.validate_python(raw)  # [1, 2, 3]

批量 ID 查询端点

from fastapi import HTTPException

@router.post("/products/batch")
async def batch_products(
    body: list[int],
    db: AsyncSession = Depends(get_db),
):
    ta = TypeAdapter(list[Annotated[int, Field(gt=0)]])
    try:
        ids = ta.validate_python(body)
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=e.errors())

    products = await get_products_by_ids(db, ids)
    return {"code": 0, "message": "ok", "data": products}

校验嵌套 JSON 字符串（第三方回调）

from pydantic import TypeAdapter

CallbackPayload = TypeAdapter(dict[str, ProductCreate])

def parse_callback(raw: str) -> dict[str, ProductCreate]:
    import json
    data = json.loads(raw)
    return CallbackPayload.validate_python(data)

TypeAdapter vs BaseModel