Python 类型系统与 Pydantic V2:从注解到极致运行时验证

Python 3.9-3.12 类型系统全景:内置类型、typing 模块演进、Generic/Protocol/TypeAliasType、Pydantic V2 模型定义/字段校验/序列化/Settings、与 mypy/pyright 的集成策略。包含大量可运行代码示例。

渐进式类型是 Python 的骄傲——你可以从「零注解」的脚本平滑过渡到「全注解」的工程代码。本文覆盖完整路径。


1. Python 类型注解基础

1.1 内置类型(Python 3.9+ 语法)

Python 3.9 起,内置集合类型支持泛型语法,无需导入 typing

# Python 3.9+
def process_items(items: list[int]) -> dict[str, int]:
    return {f"item_{i}": i for i in items}

# 可选参数
def greet(name: str, greeting: str | None = None) -> str:
    return f"{greeting or 'Hello'}, {name}!"

类型对照表:

Python 3.8 及以前Python 3.9+含义
typing.List[int]list[int]整数列表
typing.Dict[str, int]dict[str, int]字符串→整数映射
typing.Set[str]set[str]字符串集合
typing.Tuple[int, str]tuple[int, str]固定长度元组
typing.Optional[int]int | None可选类型(3.10+)
typing.Union[int, str]int | str联合类型(3.10+)

1.2 函数签名类型

from collections.abc import Callable, Iterator
from typing import ParamSpec, TypeVar

T = TypeVar('T')
P = ParamSpec('P')

def cache(func: Callable[P, T]) -> Callable[P, T]:
    """带缓存的装饰器,完整类型签名"""
    memo: dict = {}
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
        key = (args, tuple(kwargs.items()))
        if key not in memo:
            memo[key] = func(*args, **kwargs)
        return memo[key]
    return wrapper

@cache
def fib(n: int) -> int:
    return n if n < 2 else fib(n - 1) + fib(n - 2)

2. 高级类型系统

2.1 TypeVar(类型变量)

from typing import TypeVar

T = TypeVar('T')
T_co = TypeVar('T_co', covariant=True)  # 协变
T_contra = TypeVar('T_contra', contravariant=True)  # 逆变

# 约束 TypeVar
Number = TypeVar('Number', int, float)

def add(a: Number, b: Number) -> Number:
    return a + b  # 只能是 int 或 float

2.2 Generic(泛型类)

from typing import Generic, TypeVar

T = TypeVar('T')

class Stack(Generic[T]):
    def __init__(self) -> None:
        self._items: list[T] = []
    
    def push(self, item: T) -> None:
        self._items.append(item)
    
    def pop(self) -> T:
        return self._items.pop()
    
    def peek(self) -> T | None:
        return self._items[-1] if self._items else None

# 使用
int_stack: Stack[int] = Stack()
int_stack.push(42)
# int_stack.push("x")  # 💥 mypy 报错!

2.3 Protocol(结构子类型)

Python 的 Protocol == TypeScript 的 Interface == Rust 的 Trait(结构层面):

from typing import Protocol

class Drawable(Protocol):
    def draw(self) -> None: ...

class Circle:
    def draw(self) -> None:
        print("Drawing circle")

class Square:
    def draw(self) -> None:
        print("Drawing square")

def render(items: list[Drawable]) -> None:
    for item in items:
        item.draw()

# ✅ Circle 和 Square 都没有「继承」Drawable,但结构兼容
render([Circle(), Square()])

2.4 TypedDict(结构化字典)

from typing import TypedDict, NotRequired  # Python 3.11+

class Movie(TypedDict):
    name: str
    year: int
    rating: NotRequired[float]

movie: Movie = {"name": "Inception", "year": 2010}
# movie["rating"]  # 类型检查器知道这是 float | None

2.5 类型别名与 NewType

from typing import NewType

# .type 别名(3.10+)
Vector = list[float]
Matrix = list[Vector]

# NewType:创建语义不同的名义类型
UserId = NewType('UserId', int)
OrderId = NewType('OrderId', int)

def fetch_user(user_id: UserId) -> dict:
    ...

# fetch_user(OrderId(123))  # 💥 mypy 报错,语义不同

3. Pydantic V2:运行时类型验证

3.1 为什么需要 Pydantic

Python 的类型注解仅在静态检查时生效。Pydantic 在运行时验证数据:

from pydantic import BaseModel, Field, ValidationError

class User(BaseModel):
    id: int
    name: str = Field(min_length=1, max_length=50)
    email: str = Field(pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    age: int = Field(ge=0, le=150, default=0)
    is_active: bool = True

# ✅ 有效数据
user = User(id=1, name="Alice", email="alice@example.com", age=30)
print(user.model_dump())
# {'id': 1, 'name': 'Alice', 'email': 'alice@example.com', 'age': 30, 'is_active': True}

# ❌ 无效数据 → 抛出 ValidationError
try:
    User(id="not_int", name="", email="bad-email")
except ValidationError as e:
    print(e)
    # 3 validation errors...

3.2 字段校验详解

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Annotated
from datetime import datetime

class Order(BaseModel):
    order_id: str = Field(pattern=r'^ORD-\d{6}$')
    items: list[str] = Field(min_length=1)
    price: float = Field(gt=0)
    created_at: datetime = Field(default_factory=datetime.now)
    
    # 自定义字段校验
    @field_validator('items', mode='before')
    @classmethod
    def split_csv(cls, v):
        if isinstance(v, str):
            return [item.strip() for item in v.split(',')]
        return v
    
    # 模型级校验
    @model_validator(mode='after')
    def check_total(self):
        if len(self.items) > 10 and self.price < 100:
            raise ValueError('批量订单最低金额 ¥100')
        return self

# CSV 字符串自动拆分
order = Order(order_id="ORD-123456", items="apple, banana, cherry", price=150.0)
print(order.items)  # ['apple', 'banana', 'cherry']

3.3 复杂模型:嵌套与继承

from pydantic import BaseModel
from typing import Literal

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

class Payment(BaseModel):
    method: Literal['credit_card', 'alipay', 'wechat']
    amount: float

class CustomerOrder(BaseModel):
    customer_id: int
    address: Address
    payment: Payment
    items: list[str]
    
    def summary(self) -> str:
        return f"Order for {self.customer_id}: {len(self.items)} items, ¥{self.payment.amount}"

# 从 JSON/dict 直接构造
data = {
    "customer_id": 42,
    "address": {
        "street": "科技园路 1 号",
        "city": "深圳",
        "postal_code": "518000"
    },
    "payment": {
        "method": "alipay",
        "amount": 299.99
    },
    "items": ["键盘", "鼠标"]
}
order = CustomerOrder.model_validate(data)

3.4 Config 与序列化

from pydantic import BaseModel, ConfigDict
from datetime import datetime

class Event(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,      # 自动去除字符串首尾空格
        str_to_lower=True,              # 字符串转小写
        validate_assignment=True,       # 赋值时也校验
        extra='forbid',                 # 禁止额外字段
    )
    
    name: str
    timestamp: datetime
    tags: list[str] = []

# 序列化选项
event = Event(name="  Launch  ", timestamp=datetime.now())
print(event.model_dump())                    # dict
print(event.model_dump_json(indent=2))       # JSON 字符串
print(event.model_dump(mode='json'))         # JSON 兼容 dict(datetime → ISO 字符串)

3.5 Pydantic Settings(配置管理)

from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    app_name: str = "MyApp"
    debug: bool = False
    database_url: str
    secret_key: str
    max_workers: int = 4
    
    model_config = ConfigDict(
        env_file='.env',
        env_file_encoding='utf-8',
        case_sensitive=False,
    )

@lru_cache
def get_settings() -> Settings:
    """缓存配置,避免重复读取环境变量"""
    return Settings()

# 使用
settings = get_settings()
print(settings.database_url)

4. mypy vs pyright:静态分析实战

4.1 mypy

pip install mypy
mypy src/ --strict --show-error-codes

配置 pyproject.toml

[tool.mypy]
python_version = "3.11"
strict = true
warn_return_any = true
warn_unused_ignores = true
disallow_untyped_defs = true
ignore_missing_imports = true

4.2 pyright / Pylance

VS Code 内置 Pylance 即 pyright 的超集:

pip install pyright
pyright src/

配置 pyproject.toml

[tool.pyright]
pythonVersion = "3.11"
typeCheckingMode = "strict"
include = ["src"]
exclude = ["**/test_*"]

4.3 两者对比

维度mypypyright
检查速度较慢(增量检查)较快(语言服务器)
错误信息详细简洁
与 IDE 集成需要插件Pylance 原生
类型推断保守更激进
建议CI 使用开发时使用

5. 完整实战:API 请求模型

from pydantic import BaseModel, Field, HttpUrl, validator
from datetime import datetime
from typing import Literal

class PaginationParams(BaseModel):
    page: int = Field(1, ge=1)
    page_size: int = Field(20, ge=1, le=100)
    
    @property
    def offset(self) -> int:
        return (self.page - 1) * self.page_size

class CreateUserRequest(BaseModel):
    username: str = Field(min_length=3, max_length=20, pattern=r'^[a-zA-Z0-9_]+$')
    email: str  # Pydantic 内置 email 校验(需 email-validator)
    role: Literal['admin', 'user', 'guest'] = 'user'
    bio: str | None = Field(None, max_length=500)

class APIResponse(BaseModel):
    success: bool
    data: list[BaseModel] | BaseModel | None = None
    message: str | None = None
    timestamp: datetime = Field(default_factory=datetime.utcnow)

# 使用
from fastapi import FastAPI

app = FastAPI()

@app.post("/users", response_model=APIResponse)
async def create_user(req: CreateUserRequest):
    # req 已经是校验过的 Pydantic 对象
    return APIResponse(success=True, data=req)

6. 常见模式速查

场景方案
可选字段field: int | None = NoneNotRequired
默认值工厂Field(default_factory=list)
字段别名Field(alias="userName")
JSON Schemamodel_json_schema()
部分更新model_copy(update={...})
ORM 集成model_config = ConfigDict(from_attributes=True)

延伸阅读

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「python」更多文章