Python · 闭包、装饰器与 functools

场景 1 · 三层装饰器栈: 审计 + 重试守支付回调

支付回调入口需要"耗时审计 + 瞬时错误重试"两个横切能力, 不想侵入业务代码, 用装饰器栈组装:

import functools, time, logging

def audit(fn):                              # 最外层: 记进入/退出与耗时
    @functools.wraps(fn)
    def wrapper(*a, **kw):
        t0 = time.monotonic()
        try:
            return fn(*a, **kw)             # 返回值必须透传, 不能吞
        finally:
            logging.info("%s %.1fms", fn.__name__,
                         (time.monotonic() - t0) * 1e3)
    return wrapper

@audit                    # 自下而上应用: retry 先包住业务
@retry(times=3)           # retry 工厂见场景 9
def call_payment(oid):
    return gateway.charge(oid)

调用链: audit → retry → charge; charge 抛 TransientError 时在 retry 层消化重试, 审计层只看到最终耗时 —— 每层职责单一, 叠放顺序即语义。

场景 2 · lru_cache 扛住商品元数据读放大

列表页一次请求要读几百个 SKU 元数据, 重复 SKU 与跨请求重复打到 DB, 读 QPS 3200; 元数据是只读热数据, 正是 lru_cache 的主场:

from functools import lru_cache

@lru_cache(maxsize=4096)                   # 有上界: 防无界增长
def sku_meta(sku_id: int) -> dict:
    return db.one("SELECT price,title FROM sku WHERE id=%s", (sku_id,))

def render_page(skus):
    return [sku_meta(s) for s in skus]     # 重复 SKU 直接命中

# 上线后: 元数据 DB QPS 3200 → 约 100 (命中率 97%), 列表页 P99 210ms → 18ms
assert sku_meta.cache_info().hits > 0      # cache_info 必须接监控

场景 3 · wraps 救活 FastAPI 的参数反射

团队统一的 traced 装饰器上线后, 所有路由的 OpenAPI schema 塌成 (*args, **kwargs), 前端批量 422; 缺的就是 @wraps:

import functools, inspect

def traced(fn):
    @functools.wraps(fn)                   # 复制 __name__/__doc__/__wrapped__/签名
    def wrapper(*args, **kw):
        return fn(*args, **kw)
    return wrapper

@traced
def create_user(name: str, age: int = 18): ...

inspect.signature(create_user)             # → (name: str, age: int = 18) ✓
# 没有 wraps 时 __name__=='wrapper', FastAPI 读不到注解, 
# 依赖注入/文档/序列化全瞎 —— 一个装饰器行引发的线上事故

场景 4 · partial 绑定重复参数做回调注册表

事件分发要按 topic 路由到"频道 + 事件名"组合, 不想写一堆胶水函数, 用 partial 预绑定前两个参数:

from functools import partial

def send(channel: str, event: str, payload: dict): ...

ROUTES = {                                 # 注册表: 预绑定固定的两个参数
    "order.paid":     partial(send, "email", "order_paid"),
    "order.refunded": partial(send, "sms",   "order_refund"),
    "user.signup":    partial(send, "push",  "welcome"),
}

def dispatch(topic: str, payload: dict):
    ROUTES[topic](payload)                 # 新事件 = 注册表加一行, dispatch 永不改

场景 5 · 闭包实现进程内限流器 (nonlocal 正确用法)

单实例 QPS 限流不想引 Redis, 用闭包把窗口状态藏在函数里, 零类零全局变量:

import time

def rate_limiter(max_qps: int):
    window, count = 0.0, 0                  # 状态活在 cell 里
    def allow() -> bool:
        nonlocal count, window             # 声明: 改的是外层 cell
        now = time.monotonic()
        if now - window >= 1.0:
            window, count = now, 0         # 滚进新的一秒窗口
        count += 1
        return count <= max_qps
    return allow

allow = rate_limiter(100)                   # 每个 limiter 独立一套 cell

两个 limiter 互不干扰 —— make 每调用一次就造一组新 cell, 这正是"工厂函数"的价值。

场景 6 · 延迟绑定事故: 三个告警回调全重启了 mq

巡检脚本给 db/cache/mq 三个组件各注册一个重启回调, 告警触发时三个回调全重启了最后一个组件 mq:

services = ["db", "cache", "mq"]
# 事故代码: 三个 lambda 共享同一个 svc 的 cell, 取值在调用时
# handlers = [lambda: restart(svc) for svc in services]
# handlers[0]() → restart("mq") —— 三个回调全重启了最后一个 ✗

# 修法一: 默认参数在 def 时求值, 把当前值钉住
handlers = [lambda svc=svc: restart(svc) for svc in services]

# 修法二: 工厂函数, 每次调用造独立 cell
def bind(svc):
    def _restart(): return restart(svc)
    return _restart
handlers = [bind(s) for s in services]     # handlers[0]() 正确重启 db ✓

场景 7 · singledispatch 按 payload 版本分派 handler

同一 topic 的消息体有 V1/V2 两代结构, if isinstance 分支越积越多; 换成类型分派, 新版本=新注册:

from functools import singledispatch

@singledispatch
def render(payload) -> dict:               # 默认兜底分支必须有
    return {"type": "unknown"}

@render.register
def _(p: V2Payload) -> dict:               # 按第一个参数的注解分派
    return {"v": 2, **p.model_dump()}

@render.register
def _(p: V1Payload) -> dict:
    return {"v": 1, **migrate_v1(p)}    # 老版本走兼容分支

def on_request(req):
    return render(parse_body(req))         # 干掉 if isinstance 长链

场景 8 · cached_property 惰性求值昂贵报表

报表对象的 rows 要跑一次 1.8s 的多表 join, 但九成请求只用到摘要字段, 把 rows 做成惰性缓存:

from functools import cached_property

class MemberReport:
    def __init__(self, uid: int):
        self.uid = uid

    @cached_property
    def rows(self) -> list:                # 首次访问才执行, 之后读实例 __dict__
        return heavy_join(self.uid)

r = MemberReport(42)
r.rows is r.rows                         # True: 第二次不再进描述符
# 注意: 实例常驻 (如 session 级缓存) 时数据会陈旧,
# 主动失效: del r.rows 之后又会走一次真正的计算

摘要类请求 P99 从 1.9s 回到 30ms; 全量请求同实例只 join 一次。

场景 9 · 带参数的重试装饰器工厂 (三层嵌套)

重试次数和退避基数要按接口配置, 装饰器必须能收参数 —— 这就是"工厂 → 装饰器 → wrapper"三层嵌套:

import functools, time

def retry(times=3, backoff=0.2):           # 三层: 工厂(参数) → deco(函数) → wrapper
    def deco(fn):
        @functools.wraps(fn)
        def wrapper(*a, **kw):
            last = None
            for i in range(times):
                try:
                    return fn(*a, **kw)
                except TransientError as e:    # 捕窄: 业务错误不重试
                    last = e; time.sleep(backoff * 2 ** i)  # 指数退避
            raise last
        return wrapper
    return deco

@retry(times=5, backoff=0.1)               # 先调用 retry(5, 0.1) 拿到真装饰器
def sync_inventory(): ...

场景 10 · 装饰器做依赖注入: 把 session 织进函数参数

业务函数都想要一个 DB session, 但手动管理 open/commit/rollback 到处重复; 用装饰器按需注入, 外层事务还能透传:

import functools

def with_session(fn):
    @functools.wraps(fn)
    def wrapper(*args, session=None, **kw):
        if session is not None:            # 调用方传了: 加入外层事务
            return fn(*args, session=session, **kw)
        with Session() as s:               # 没传: 这里管生命周期
            return fn(*args, session=s, **kw)
    return wrapper

@with_session
def transfer(src, dst, amount, session): ...

transfer(1, 2, 99.0)                       # 自动开/提交/回滚
transfer(1, 2, 99.0, session=outer_tx)     # 挂进外层事务 —— 测试替身免费