Tiana随笔

Python 装饰器:把重复逻辑收起来

· Python

标签:#Python #backend

Python 装饰器:把重复逻辑收起来

第一次写装饰器我漏了 wraps,函数的名字和文档字符串全变成 wrapper 的。后面记计时、权限校验、缓存、重试四个真在用的装饰器,以及它们各自在哪儿翻过车。

我第一次用装饰器就范的错误

最早期的 Python 项目里我写过这么一段:

def timer(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print(f"cost: {time.time() - start}")
        return result
    return wrapper

@timer
def query_user(user_id):
    return db.get(User, user_id)

print(query_user.__name__)
# 输出: wrapper

我盯着「wrapper」这个字符串 debug 一晚上没查出来为啥 unit test 失败,原因是 traceback 显示是 wrapper 报的错。,我一脸懵:我的函数叫 query_user 啊。

那晚学到一件事:@functools.wraps 不是可选项。

@wraps 一行解决属性继承

from functools import wraps

def timer(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print(f"cost: {time.time() - start}")
        return result
    return wrapper

@wraps(func) 这一行做的事情是:把 func.__name__、func.__doc__、func.__module__、func.__dict__ 等一堆属性复制到 wrapper。debug 工具、Sphinx 自动文档、Flask 路由元信息查询,全都依赖这些属性。

装饰器里不加 @wraps 是新人最常见的坑。看到项目里 @wraps 没加你就知道是别人没注意到这一层。

真正干活:计时装饰器

我自己常用的版本,多带几个细节:

import time
import logging
from functools import wraps

log = logging.getLogger(__name__)

def timer(label: str | None = None, threshold_ms: int = 100):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                return func(*args, **kwargs)
            finally:
                cost_ms = (time.perf_counter() - start) * 1000
                if cost_ms > threshold_ms:
                    log.warning("slow_call",
                        extra={"func": func.__name__, "label": label, "cost_ms": cost_ms})
        return wrapper
    return decorator

挂的时候:

@timer("search_index", threshold_ms=200)
def rebuild_index():
    index.clear()
    for post in posts.all():
        index.add(post)

这里我用了「工厂装饰器」模式:外层函数接收参数,内层才是真装饰器。不接参数的版本简单到 @timer 一行:外层就是装饰器本身。

工厂模式最大的好处是能在挂载时配行为。比如 threshold_ms 控制多少毫秒才报警,不重要的函数不想打扰可以挂 @timer() 走默认阈值。

装饰器做权限校验

后端开发里最常见的装饰器用途其实是权限校验。

def require_admin(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        sess = kwargs.get("session") or current_session()
        if not sess or sess.role != "admin":
            raise HTTPException(401, detail={"code": "AUTH_REQUIRED"})
        return func(*args, **kwargs)
    return wrapper

挂法:

@router.post("/admin/backup/create")
@require_admin
def create_backup(session=Depends(get_session)):
    return BackupService.create()

这种写法在 FastAPI 里非常顺手。整个后台接口用同一个装饰器挂着,谁忘挂哪个一目了然。同样的思路可以扩展到 require_auth、require_role("editor")、rate_limit(10) 等。

装饰器做缓存

进程内带 TTL 缓存是个非常常见的诉求,模式是「外层 dict 拿 key,外面圈一层时间戳判过期,更新值」。具体写法这里不展开,搜「Python lru_cache with TTL」能找到很多变体。但有一点要说清楚:进程内缓存进程重启就丢,多实例部署时还各自一份。要 robust 就用 cachetools.TTLCache,或者干脆上 Redis。

重试装饰器:业务侧打补丁

外部接口偶尔抖动时常常需要重试,最核心的逻辑是 try / except 加循环 + 退出条件。base_delay 2 * n 是退避因子(第一次等一个 base、再倍增),加一点 random.random() 避免所有重试同步打过去。比装饰器更常见的做法其实是框架自带的 retry middleware,挂上去就行;如果要写,模式基本一致。

装饰器能干啥不能干啥

我给自己定了几条规矩:

做

  • 计时、日志、埋点
  • 权限校验、限流
  • 输入输出格式化
  • 缓存、重试、超时控制
  • 上下文注入(session、request id、trace)

慎做

  • 业务逻辑(应该改 service 层而不是装饰器)
  • 副作用重的操作(reduce 之类的 hook 不在装饰器里悄悄做)
  • 装饰器链超过 4 层(可读性崩溃)

理由是装饰器把代码流「藏」起来。每多用一层,新人 debug 时就要多想一层。装饰器只该用在横切关注点上,业务逻辑直接调用比套装饰器清楚得多。

几个真实翻车点

装饰器顺序有讲究。@timer 和 @retry 同时挂,怎么决定谁在内谁在外?

@timer
@retry(max_attempts=3)
def call_remote():
    ...

这写法 timer 包了 retry,所以计时包含所有重试。如果想只看「最后一次成功调用」的耗时:

@retry(max_attempts=3)
@timer
def call_remote():
    ...

timer 在内,每次重试都重新计时。

装饰器装饰器叠太多。我见过一个函数挂了 5 层装饰器,bug 排查时只看函数签名能猜出大概?根本猜不出。这就是我说的「最多 4 层」的理由。

装饰器自己写太复杂。如果装饰器函数比被装饰的函数还长,多半是写错了。把那段抽成普通函数,再让装饰器去调用。

写装饰器的最后一步:测试

我自己写完装饰器先写两个测试。一个验证 @wraps 真的把原函数名复制了,另一个验证装饰器真的跑了被装饰函数(写这种测试的人容易只顾装饰器本体,忘了被装饰函数的契约会变化):

def test_wraps_preserves_name():
    @timer
    def sample(): pass
    assert sample.__name__ == "sample"

def test_timer_runs_once():
    calls = []
    @timer
    def sample():
        calls.append(1); return "ok"
    assert sample() == "ok" and calls == [1]

相关文章