Pydantic v2 与 FastAPI：验证、序列化与类型安全

Pydantic v2：有哪些变化

Pydantic v2 用 Rust 重写——验证速度提升 5-50 倍。此外还新增了计算字段、模型验证器和判别联合等功能。

基础模型

from pydantic import BaseModel, Field, EmailStr, model_validator, computed_field
from typing import Annotated
from datetime import datetime

class UserCreate(BaseModel):
    name: Annotated[str, Field(min_length=2, max_length=100)]
    email: EmailStr
    password: Annotated[str, Field(min_length=8)]
    age: Annotated[int, Field(ge=0, le=150)] | None = None

class UserResponse(BaseModel):
    id: str
    name: str
    email: str
    created_at: datetime

    @computed_field
    @property
    def display_name(self) -> str:
        return f"{self.name} ({self.email})"

    model_config = {
        "from_attributes": True,  # 启用 ORM 模式
    }

模型验证器

from pydantic import model_validator, field_validator

class PasswordChange(BaseModel):
    current_password: str
    new_password: str
    confirm_password: str

    @model_validator(mode='after')
    def passwords_must_match(self) -> 'PasswordChange':
        if self.new_password != self.confirm_password:
            raise ValueError('Passwords do not match')
        if self.new_password == self.current_password:
            raise ValueError('New password must differ from current')
        return self

class UserProfile(BaseModel):
    username: str
    website: str | None = None

    @field_validator('username')
    @classmethod
    def username_alphanumeric(cls, v: str) -> str:
        if not v.replace('_', '').isalnum():
            raise ValueError('Username must be alphanumeric (underscores allowed)')
        return v.lower()

    @field_validator('website')
    @classmethod
    def website_valid_url(cls, v: str | None) -> str | None:
        if v and not v.startswith(('http://', 'https://')):
            return f'https://{v}'
        return v

判别联合

from typing import Literal, Union
from pydantic import BaseModel, Field

class EmailNotification(BaseModel):
    type: Literal['email']
    to: EmailStr
    subject: str
    body: str

class SmsNotification(BaseModel):
    type: Literal['sms']
    phone: str
    message: str

class PushNotification(BaseModel):
    type: Literal['push']
    device_token: str
    title: str
    body: str

Notification = Annotated[
    Union[EmailNotification, SmsNotification, PushNotification],
    Field(discriminator='type')
]

class NotificationRequest(BaseModel):
    notification: Notification
    schedule_at: datetime | None = None

# FastAPI endpoint
@router.post('/notifications')
async def send_notification(req: NotificationRequest):
    match req.notification:
        case EmailNotification(to=email, subject=subj, body=body):
            await email_service.send(to=email, subject=subj, body=body)
        case SmsNotification(phone=phone, message=msg):
            await sms_service.send(phone=phone, message=msg)
        case PushNotification(device_token=token, title=title, body=body):
            await push_service.send(token=token, title=title, body=body)

使用 pydantic-settings 管理设置

from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file='.env',
        env_file_encoding='utf-8',
        case_sensitive=False,
    )

    # App
    app_name: str = 'MyAPI'
    debug: bool = False
    secret_key: str

    # Database
    database_url: str
    db_pool_size: int = 10
    db_max_overflow: int = 20

    # Redis
    redis_url: str = 'redis://localhost:6379'
    redis_ttl: int = 3600

    # External services
    stripe_api_key: str | None = None
    sendgrid_api_key: str | None = None

@lru_cache
def get_settings() -> Settings:
    return Settings()

自定义序列化

from pydantic import BaseModel
from pydantic.functional_serializers import model_serializer, field_serializer
from decimal import Decimal

class Money(BaseModel):
    amount: Decimal
    currency: str = 'USD'

    @field_serializer('amount')
    def serialize_amount(self, value: Decimal) -> str:
        return str(value.quantize(Decimal('0.01')))

class Order(BaseModel):
    id: str
    items: list[dict]
    total: Money
    created_at: datetime

    @model_serializer(mode='wrap')
    def custom_serializer(self, handler) -> dict:
        data = handler(self)
        data['_links'] = {
            'self': f'/orders/{self.id}',
            'items': f'/orders/{self.id}/items',
        }
        return data

FastAPI 集成技巧

# 使用 response_model 过滤输出字段
@router.get('/users/{id}', response_model=UserResponse)
async def get_user(id: str, db: AsyncSession = Depends(get_db)):
    user = await db.get(User, id)
    if not user:
        raise HTTPException(404)
    return user  # Pydantic 自动转换 ORM 模型

# 使用 Annotated 实现可复用的验证
UserId = Annotated[str, Path(pattern=r'^[0-9a-f-]{36}
#39;)]
PageSize = Annotated[int, Query(ge=1, le=100, default=20)]

@router.get('/users')
async def list_users(page: int = Query(default=1, ge=1), size: PageSize = 20):
    ...