Python · 描述符与装饰器 (元编程)

场景 1 · 重试装饰器 (调下游的标配)

def retry(times=3, delay=0.5, exc=(TimeoutError, ConnectionError)):
    def deco(fn):
        @functools.wraps(fn)                     # 保住函数名/签名/文档
        def wrapper(*args, **kw):
            for i in range(times, 0, -1):
                try: return fn(*args, **kw)
                except exc as e:
                    if i == 1: raise
                    time.sleep(delay * (times - i + 1))   # 递增退避
        return wrapper
    return deco

@retry(times=3)
def call_payment(api): ...                     # 幂等接口才可重试!

注意重试的前提是幂等; 支付下单这类非幂等操作要用幂等键, 而不是裸重试。

场景 2 · 类型校验描述符 (ORM 风格基建)

配置/模型字段上线前就拦住脏数据, 而不是等线上报错:

class Typed:
    def __init__(self, typ, default=None): self.typ, self.name = typ, None
    def __set_name__(self, owner, name): self.name = name    # 3.6+: 自动拿字段名
    def __set__(self, obj, val):
        if not isinstance(val, self.typ):
            raise TypeError(f"{self.name} 需要 {self.typ.__name__}, 得到 {val!r}")
        obj.__dict__[self.name] = val          # 值仍存实例字典, 描述符只做守门
    __get__ = None  # 简化: 读直接走实例字典 (跳过 __get__)

class Order:
    amount = Typed((int, float));  status = Typed(str)

Order().amount = "1"   # → TypeError: amount 需要 int/float, 得到 '1'

场景 3 · property 做派生字段 + 缓存

class Sku:
    @property
    def price_with_tax(self):              # 读起来像字段, 每次实时算
        return round(self.price * (1 + TAX), 2)

    @functools.cached_property            # 3.8+: 算一次缓存进 __dict__
    def heavy_report(self): ...

接口序列化时 property 直接当字段输出 (dataclasses.asdict 前先 asdict() 转换), 派生字段不必落库。

场景 4 · 路由注册装饰器(框架的骨架)

Flask 风格 @app.get 的本质: 装饰器把 (path, handler) 登记进注册表并原样返回函数:

ROUTES: dict[str, callable] = {}
def route(path):
    def deco(fn):
        ROUTES[path] = fn                  # 注册表登记
        @functools.wraps(fn)
        def wrapper(*a, **kw):
            return fn(*a, **kw)            # 原样透传, 也可加前置/后置
        return wrapper
    return deco

@route("/orders")                       # 业务函数零侵入接入
def list_orders(): ...

场景 5 · 依赖注入描述符

服务对象的懒初始化与共享用描述符最顺: 业务类只见声明, 组装逻辑集中在容器:

class Inject:
    def __init__(self, cls): self.cls = cls
    def __set_name__(self, owner, name): self.name = name
    def __get__(self, obj, owner=None):
        if obj is None: return self
        svc = container.resolve(self.cls)          # 首次访问时按类型解析
        obj.__dict__[self.name] = svc              # 缓存进实例, 之后直读
        return svc

class OrderService:
    db = Inject(Database)          # 声明即注入; 单测时 container.override(Database, FakeDB)

场景 6 · 强类型配置类(pydantic 思想)

配置字段赋值即校验, 错误在启动第 1 秒爆出而不是运行第 3 天:

class Range:
    def __init__(self, typ, lo, hi): self.typ, self.lo, self.hi = typ, lo, hi
    def __set_name__(self, o, name): self.name = name
    def __set__(self, obj, val):
        if not isinstance(val, self.typ) or not (self.lo <= val <= self.hi):
            raise ValueError(f"{self.name} 需 {self.typ.__name__} 且在 [{self.lo},{self.hi}]")
        obj.__dict__[self.name] = val

class AppConfig:
    port = Range(int, 1, 65535)
    workers = Range(int, 1, 128)
AppConfig().port = 99999   # → ValueError: 启动即失败

场景 7 · 大对象惰性派生字段

报告生成贵且多数请求用不到: cached_property 延迟到首次访问, 失效可精确控制:

class SkuView:
    def __init__(self, sku): self.sku = sku
    @functools.cached_property              # 算一次缓存进实例 __dict__
    def heavy_report(self):
        return build_expensive_report(self.sku)
    def invalidate(self):
        self.__dict__.pop("heavy_report", None)   # 删缓存即重新生成

场景 8 · ORM 关系字段与 N+1 查询

order.user 首次访问触发查询并缓存 —— 看懂描述符就懂 N+1 的成因与解法:

class ForeignKey:
    def __init__(self, model): self.model = model
    def __set_name__(self, o, name): self.name = name
    def __get__(self, obj, owner=None):
        if obj is None: return self
        if self.name not in obj._cache:
            obj._cache[self.name] = self.model.get(obj._fk[self.name])  # 惰性查询!
        return obj._cache[self.name]
# 循环里逐个 order.user = 每次一条 SQL = N+1; 解法: select_related 一次预填 _cache

场景 9 · 认证与授权的装饰器编排

请求从上往下穿: 先认证(身份)再授权(权限), 顺序即语义:

def login_required(fn):
    @functools.wraps(fn)
    def wrapper(request, *a, **kw):
        if not request.user: raise Unauthorized()
        return fn(request, *a, **kw)
    return wrapper

def role_required(role):
    def deco(fn):
        @functools.wraps(fn)
        def wrapper(request, *a, **kw):
            if request.user.role != role: raise Forbidden()
            return fn(request, *a, **kw)
        return wrapper
    return deco

@login_required                      # 外层: 有身份
@role_required("admin")             # 内层: 是管理员
def delete_order(request): ...

场景 10 · 插件自动注册体系

基类 __init_subclass__(3.6+) 子类定义即注册, 不需要 metaclass:

REGISTRY: dict[str, type] = {}
class Plugin:
    name: str
    def __init_subclass__(cls, **kw):          # 每定义一个子类自动调用
        super().__init_subclass__(**kw)
        REGISTRY[cls.name] = cls             # 注册!无需显式登记

class WechatPay(Plugin): name = "wechat"    # import 即生效
class Alipay(Plugin):  name = "alipay"
handler = REGISTRY["wechat"](cfg)             # 按名实例化, 新增支付方式零改分发代码