Python · 类型注解与静态检查

场景 1 · 30 万行存量库渐进接入 mypy, 不停业务迭代

一次性全量 strict 等于一次报出五万个错, 没人敢合。正解是配置分层: 新模块最严, 旧模块先豁免再逐步收网:

# pyproject.toml — 渐进式收网: 新代码 strict, 旧代码先放行
[tool.mypy]
python_version = "3.11"
plugins = ["pydantic.mypy"]          # 不装插件: pydantic 字段在 mypy 眼里全是 Any
warn_unused_ignores = true           # ignore 失效即告警, 防止永久遮蔽真错

[[tool.mypy.overrides]]
module = "legacy.*"                  # 存量代码: 先整体豁免, 按目录逐个摘掉
ignore_errors = true

[[tool.mypy.overrides]]
module = "app.*"                     # 新模块一律最严
disallow_untyped_defs = true
disallow_any_generics = true

CI 里 mypy 失败即红灯。三个月后 legacy 豁免名单从 47 个目录收到 6 个, 每次摘掉一个都有静态检查兜底。

场景 2 · pydantic 定义 API 请求/响应模型, FastAPI 自动 422

校验逻辑散在视图函数里既难审又难测。把契约写成模型, 校验、文档、错误格式一次到位:

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

class TransferRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")     # 未知字段直接拒: 防客户端塞脏参数
    src: str = Field(min_length=8, max_length=8)  # 约束跟类型写在一起, 一眼可审
    amount: Decimal = Field(gt=Decimal("0"), max_digits=12)

class TransferResponse(BaseModel):
    receipt_no: str
    balance: Decimal

@app.post("/transfer", response_model=TransferResponse)
def transfer(req: TransferRequest):               # 进函数体时已是校验过的强类型对象
    ...
# amount=-1 或多余字段: 自动 422 + 字段级报错, OpenAPI 文档同步生成

场景 3 · Protocol 解耦第三方资源, 老类一行不改

想让函数接受"任何可读可关的资源"。继承抽象基类要改别人代码, Protocol 只认形状:

from typing import Protocol

class Readable(Protocol):          # 结构化类型: 只看方法形状, 不要求继承
    def read(self, n: int) -> bytes: ...
    def close(self) -> None: ...

def fetch_header(res: Readable):    # 文件/socket/自研连接/测试替身都能传
    try:
        return res.read(16)
    finally:
        res.close()

class LegacyFTP:                     # 十年前老类, 零改动即满足 Readable
    def read(self, n: int) -> bytes: ...
    def close(self) -> None: ...

场景 4 · TypedDict 约束 JSON 配置文件

配置是 dict 不是类, 但字段名打错、类型给错这类问题完全可以让静态检查器先抓一遍:

from typing import TypedDict, Literal
import json, pathlib

class DBConfig(TypedDict):
    host: str
    port: int
    engine: Literal["mysql", "postgres"]   # 写 "mongo" 直接红线

class AppConfig(TypedDict, total=False):    # total=False: 所有键都可选
    db: DBConfig
    debug: bool

cfg: AppConfig = json.loads(pathlib.Path("app.json").read_text())
cfg["port"] = "3306"      # mypy 立刻报: str 塞不进 int 槽位
# 注意: TypedDict 运行时不校验 — 配置来源不可信时要换 pydantic 解析

场景 5 · Literal + Enum 收窄状态机, 配 match 穷尽检查

状态值用裸字符串, 新增状态后总有分支被漏。枚举 + assert_never 让"漏写分支"变成静态错误:

import enum
from typing import assert_never

class OrderStatus(str, enum.Enum):
    PAID = "paid"
    SHIPPED = "shipped"
    DONE = "done"

def next_action(s: OrderStatus) -> str:
    match s:
        case OrderStatus.PAID:    return "notify_warehouse"
        case OrderStatus.SHIPPED: return "watch_carrier"
        case OrderStatus.DONE:    return "archive"
        case _:
            assert_never(s)   # 明天加 REFUNDED 忘了写分支: mypy 当场报错

场景 6 · 泛型仓储类 Repository[T] 一套代码全实体复用

User 仓储、Order 仓储除了实体类型什么都不差。泛型把"进出同型"写进合同:

from typing import Generic, TypeVar

T = TypeVar("T")

class Repository[T]:                      # 3.12 语法; 老版本写 Generic[T]
    def __init__(self, model: type[T]) -> None:
        self.model = model

    def get(self, oid: int) -> T | None: ...
    def save(self, obj: T) -> None: ...

user_repo: Repository[User] = Repository(User)
u = user_repo.get(42)          # u 推导为 User | None, 属性补全全有
user_repo.save(Order(...))     # mypy: Repository[User] 不收 Order

场景 7 · ParamSpec 让打点装饰器不丢签名

监控装饰器如果签名写 *args: Any, 被包函数的类型信息全灭。ParamSpec 原样透传:

from typing import ParamSpec, TypeVar, Callable
import time, functools

P = ParamSpec("P")
R = TypeVar("R")

def timed(fn: Callable[P, R]) -> Callable[P, R]:
    @functools.wraps(fn)
    def wrap(*args: P.args, **kwargs: P.kwargs) -> R:   # 签名原样透传
        t0 = time.perf_counter()
        try:
            return fn(*args, **kwargs)
        finally:
            metrics.observe(fn.__name__, time.perf_counter() - t0)
    return wrap
# 被 @timed 的函数: 参数/返回值类型一个不丢, IDE 补全照常

场景 8 · cast 收窄 ORM 边界 + type: ignore 的纪律

数据库驱动只会返回 Any, 放任不管 Any 会一路传染。边界处人工确认后收窄:

from typing import cast
from decimal import Decimal
from sqlalchemy import text

row = conn.execute(text("SELECT id, amount FROM orders WHERE id=:o"),
                   {"o": 42}).fetchone()
# 驱动层没有类型信息, 返回 Any — 这是 Any 入侵的第一站
oid = cast(int, row[0])          # 确认过 schema 后手动收窄, 零运行时开销
amount = cast(Decimal, row[1])

total = legacy_sum(raw)  # type: ignore[no-any-return]
# ignore 必须带错误码 + 限期修; warn_unused_ignores 会盯着它失效

场景 9 · pydantic v2 性能敏感路径: TypeAdapter 批量校验

事件管道每秒几十万条, 逐条构造模型开销可观。pydantic-core 的批量入口快一个量级:

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

class Event(BaseModel):
    model_config = ConfigDict(frozen=True)   # 不可变: 可哈希可共享, 校验后零拷贝
    user_id: int
    amount: Decimal
    tag: str | None = None

adapter = TypeAdapter(list[Event])           # 编译一次, 别在循环里建模型
events = adapter.validate_python(raw_batch)  # pydantic-core (Rust) 批量校验
# 100 万条: 逐条 Event(**x) 8.9s → 批量 2.1s; pydantic v1 同款 21s

场景 10 · 事故复盘: 注解没拦住 None, 深处 AttributeError

结账接口偶发 500, 栈指向最深处。根因是注解只是注释, 而静态通道没开、边界没校验:

# 事故现场: coupon 可能为 None, 但注解写着 Coupon, 解释器根本不查
# def apply_coupon(order, coupon): return order.total * (1 - coupon.rate)
# → AttributeError: 'NoneType' object has no attribute 'rate' (深处才炸)

# 复盘后双通道补齐:
def apply_coupon(order: Order, coupon: Coupon | None) -> Decimal:
    if coupon is None:                   # 类型收窄: 这行之后 coupon 就是 Coupon
        return order.total
    return order.total * (1 - coupon.rate)

class CouponPayload(BaseModel):        # 外部入参先过边界模型再进领域函数
    coupon_code: str | None
# 再配 CI 里的 mypy --strict: 传 None 给 Coupon 参数在合并前就红