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]