Python 动态语言里的安全带：Pydantic 用法与最佳实践

from pydantic import BaseModel, Field


class CreateUserRequest(BaseModel):
    email: str
    display_name: str = Field(min_length=1, max_length=64)
    age: int | None = Field(default=None, ge=0, le=150)

import json


def charge(amount: int) -> None:
    ...


payload = json.loads('{"amount": "100"}')
charge(payload["amount"])

from pydantic import BaseModel, Field


class ChargeRequest(BaseModel):
    amount: int = Field(strict=True, gt=0)


req = ChargeRequest.model_validate(payload)
charge(req.amount)

def charge(amount: int) -> None:
    ...


charge("100")  # mypy/pyright 可以在代码提交前提醒你

from typing import TypedDict


class ChargePayload(TypedDict):
    amount: int
    currency: str


def charge(payload: ChargePayload) -> None:
    amount: int = payload["amount"]

charge({"amount": "100", "currency": "USD"})

[tool.mypy]
python_version = "3.12"
warn_return_any = true
warn_unused_ignores = true
disallow_untyped_defs = true
no_implicit_optional = true
check_untyped_defs = true

[tool.pyright]
typeCheckingMode = "strict"
reportUnknownParameterType = true
reportUnknownVariableType = true
reportOptionalMemberAccess = true

[tool.ruff]
target-version = "py312"
line-length = 100

[tool.ruff.lint]
select = ["E", "F", "B", "I", "UP"]

ruff check .
mypy .
pyright
pytest

make check

import json
from typing import Any


payload: Any = json.loads(raw_body)
charge(payload["amount"])  # 静态检查很可能沉默

payload = json.loads(raw_body)
req = ChargeRequest.model_validate(payload)
charge(req.amount)

def create_invoice(payload: dict) -> None:
    amount = payload["amount"]
    currency = payload["currency"]
    # 继续往下调用支付系统

from decimal import Decimal
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field


class CreateInvoiceRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")

    amount: Decimal = Field(gt=0, max_digits=12, decimal_places=2)
    currency: Literal["USD", "CNY", "EUR"]
    customer_id: str = Field(min_length=1, max_length=64)


def create_invoice(payload: dict) -> None:
    req = CreateInvoiceRequest.model_validate(payload)
    # 从这里开始，业务代码面对的是 req，而不是裸 dict

from pydantic import BaseModel


class FeatureFlag(BaseModel):
    enabled: bool


print(FeatureFlag.model_validate({"enabled": "false"}))

from pydantic import BaseModel, ValidationError


class Payment(BaseModel):
    amount: int


try:
    Payment.model_validate({"amount": "100"}, strict=True)
except ValidationError as exc:
    print(exc.errors())

from pydantic import BaseModel, Field


class Payment(BaseModel):
    amount: int = Field(strict=True, gt=0)
    memo: str | None = None

from pydantic import BaseModel, ConfigDict


class InternalCommand(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")

    action: str
    retry_count: int

from datetime import datetime

from pydantic import BaseModel, Field, field_validator, model_validator


class MeetingRequest(BaseModel):
    topic: str = Field(min_length=1, max_length=128)
    participants: list[str] = Field(min_length=1, max_length=100)
    start_time: datetime
    end_time: datetime

    @field_validator("participants")
    @classmethod
    def normalize_participants(cls, value: list[str]) -> list[str]:
        cleaned = [item.strip().lower() for item in value if item.strip()]
        if not cleaned:
            raise ValueError("participants cannot be empty")
        return sorted(set(cleaned))

    @model_validator(mode="after")
    def check_time_range(self) -> "MeetingRequest":
        if self.end_time <= self.start_time:
            raise ValueError("end_time must be later than start_time")
        return self

pip install pydantic-settings

from pydantic import Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict


class AppSettings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="APP_",
        env_file=".env",
        extra="ignore",
    )

    service_name: str = "billing-api"
    debug: bool = False
    request_timeout_seconds: int = Field(default=3, ge=1, le=60)
    database_url: SecretStr


settings = AppSettings()

from pydantic import BaseModel, Field, TypeAdapter


class Event(BaseModel):
    name: str = Field(min_length=1)
    ts: int = Field(ge=0)


events_adapter = TypeAdapter(list[Event])


def handle_events(payload: object) -> list[Event]:
    return events_adapter.validate_python(payload)

from datetime import datetime
from decimal import Decimal

from pydantic import BaseModel, Field


class InvoiceView(BaseModel):
    invoice_id: str
    amount: Decimal
    created_at: datetime
    internal_note: str | None = Field(default=None, exclude=True)


view = InvoiceView(
    invoice_id="inv_001",
    amount=Decimal("99.90"),
    created_at=datetime.now(),
    internal_note="risk score: 42",
)

payload = view.model_dump(mode="json", exclude_none=True)

schema = CreateInvoiceRequest.model_json_schema()

def send_email(to: str, retry: int) -> None:
    ...

from pydantic import validate_call


@validate_call
def send_email(to: str, retry: int) -> None:
    ...

class ApiRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")

API Request Model -> Domain Command -> ORM Model -> API Response Model

from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator


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

    title: str = Field(min_length=1, max_length=128)
    priority: Literal["low", "medium", "high"] = "medium"
    assignee: str | None = Field(default=None, max_length=64)
    tags: list[str] = Field(default_factory=list, max_length=20)

    @field_validator("tags")
    @classmethod
    def normalize_tags(cls, value: list[str]) -> list[str]:
        normalized = [tag.strip().lower() for tag in value if tag.strip()]
        return sorted(set(normalized))

    @model_validator(mode="after")
    def check_high_priority_owner(self) -> "CreateTaskRequest":
        if self.priority == "high" and not self.assignee:
            raise ValueError("high priority task must have an assignee")
        return self

from datetime import datetime

from pydantic import BaseModel, ConfigDict


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

    task_id: str
    title: str
    priority: str
    assignee: str | None
    tags: list[str]
    created_at: datetime

def create_task(payload: dict) -> dict:
    req = CreateTaskRequest.model_validate(payload)

    task = task_service.create(
        title=req.title,
        priority=req.priority,
        assignee=req.assignee,
        tags=req.tags,
    )

    return TaskView.model_validate(task).model_dump(mode="json")

@startmindmap
skinparam backgroundColor #FFFFFF
skinparam defaultFontName Arial

*[#111827] <color:white><b>Pydantic 最佳实践</b></color>
**[#DBEAFE] 定位
***[#EFF6FF] 运行时验证
***[#EFF6FF] 边界契约
***[#EFF6FF] 不是静态检查
***[#EFF6FF] 不是业务逻辑
**[#DCFCE7] 输入边界
***[#F0FDF4] HTTP Request
***[#F0FDF4] MQ Message
***[#F0FDF4] 外部 API
***[#F0FDF4] LLM 输出
***[#F0FDF4] 配置环境变量
**[#FCE7F3] 模型实践
***[#FDF2F8] BaseModel
***[#FDF2F8] Field 约束
***[#FDF2F8] extra="forbid"
***[#FDF2F8] strict 关键字段
***[#FDF2F8] TypeAdapter
**[#FEF3C7] Validator 边界
***[#FFFBEB] 字段规范化
***[#FFFBEB] 跨字段约束
***[#FFFBEB] 避免副作用
***[#FFFBEB] 不查库不调 HTTP
**[#EDE9FE] 工程协作
***[#F5F3FF] JSON Schema
***[#F5F3FF] Contract Test
***[#F5F3FF] OpenAPI / AsyncAPI
***[#F5F3FF] make check
***[#F5F3FF] CI 质量门
**[#FEE2E2] 常见坑
***[#FEF2F2] 类型提示不验证
***[#FEF2F2] 迷信默认转换
***[#FEF2F2] ORM / API 混用
***[#FEF2F2] v1 / v2 混写
***[#FEF2F2] 敏感字段外泄
**[#E5E7EB] 组合拳
***[#F9FAFB] typing 写清意图
***[#F9FAFB] mypy / pyright 查类型
***[#F9FAFB] ruff 扫代码味道
***[#F9FAFB] pytest 验证行为
***[#F9FAFB] Pydantic 守住边界
@endmindmap