1. 装饰器到底在解决什么问题
先讲个真实的场景。前几年我维护过一整套内部运营后台,光类似的接口就有三四十个,早期代码写得比较随意,登录校验是这么干的:
def get_user_info(user_id): # 假设这里有权限判断,每次都要复制一遍 if not check_login(): return {"code": 401, "msg": "请先登录"} # 业务逻辑... return {"code": 0, "data": fetch_user(user_id)}每个函数开头都要粘一遍check_login()的判断,后来新增了一个功能却忘了加,线上被人直接拿着接口地址拖数据。这时候你意识到,这类和业务本身无关的逻辑——登录校验、日志记录、耗时统计、重试机制——如果能把它们从业务函数里抽出来,让函数只干自己该干的事,代码的质量会上一个台阶。
Python给了你一个非常顺手的工具来完成这件事,那就是装饰器。它是 Python 里基于"函数是一等对象"这个特性设计出来的语法糖。本质上,装饰器就是一个接收函数、返回新函数的函数,你写@decorator时,解释器做的就是func = decorator(func)这一步。
很多初学者看教程时有个误区,总把装饰器当成什么"高端魔法"。其实它一点都不玄乎。我用一句话跟组里的新人讲:如果函数是一台机器,装饰器就是你给这台机器外面套了一个控制面板——机器还是那台机器,但你在外面加了开关、仪表、报警器,甚至可以决定在某些情况下压根不启动这台机器。
这篇文章我会从最基础的写法开始,一路讲到带参数的装饰器、类装饰器、带状态的装饰器,中间穿插大量我实际踩过的坑和排查经验。不管你是刚学 Python 还没搞清楚@符号是什么意思,还是已经在项目里用过装饰器但总觉得哪里没吃透,这篇都值得花十分钟看完。
2. 先从函数本身说起
2.1 函数在 Python 里是一等公民
要理解装饰器,先得接受一个 Python 的基本事实:函数和整数、字符串、列表没什么区别,它可以被赋值给变量、被放进字典、被作为参数传给另一个函数、也可以被另一个函数返回出来。
def greet(name): return f"Hello, {name}" # 函数可以被赋值 f = greet print(f("张三")) # Hello, 张三 # 函数可以被放进容器 funcs = [greet, len, str.upper] print(funcs[2]("abc")) # ABC # 函数可以作为参数 def run(func, value): return func(value) print(run(greet, "李四")) # Hello, 李四这里的关键是最后一条——函数可以作为参数传递。既然函数能像数据一样被传来传去,一个函数接收另一个函数,再返回一个新函数,这在逻辑上就完全说得通了。装饰器的全部本质,其实就是基于这一个特性建立起来的。
我还记得第一次看到这段代码时的触动。以前写 Java 的时候,想给一个类的方法加日志,要么用 AOP 框架,要么在方法内部手动写日志逻辑。Python 把它简化成了"函数可以作为参数和返回值",这背后没有引入任何新的运行时机制,纯粹是语言特性组合出来的结果。
2.2 闭包是装饰器能工作的地基
光有函数传参还不够,装饰器还要依赖另一个概念——闭包。闭包的意思是:如果一个内部函数引用了外部函数作用域里的变量,那么这个内部函数会"记住"那个变量,即使外部函数已经执行完毕。
def outer(x): def inner(): return x * 10 return inner f = outer(5) print(f()) # 50outer(5)执行完以后,按理说它的局部变量x应该已经被销毁了。但inner引用了x,所以 Python 会把x保存在inner的__closure__里,等你调用f()的时候,inner依然能取到它。这就是闭包。
装饰器利用闭包做什么呢?当@decorator把原函数包裹起来时,新函数(就是我们写的wrapper)通过闭包"记住"了原来的函数,以及装饰器参数(如果有的话)。这样每次调用新函数的时候,wrapper 既可以用外层的配置参数,又可以调用原始的业务函数,两边都不耽误。
为了让你记得住这个逻辑,我给你一个生活化的比喻。闭包像是一个带记忆的小盒子,装饰器运行时,原函数和参数都放进了这个小盒子,包裹函数每次被调用时打开盒子取用。等你亲手写出第一个装饰器,再回头想这段,会觉得特别顺。
3. 从零手写第一个装饰器
3.1 最朴素的写法
假设我们想给某个函数增加一个功能:调用的时候在屏幕上画一条分割线。先看不用装饰器语法怎么写:
def add_divider(func): def wrapper(*args, **kwargs): print("=" * 30) return func(*args, **kwargs) return wrapper def get_user(): return {"name": "王五", "level": "admin"} get_user = add_divider(get_user) result = get_user()执行结果是先打印一行等号,然后返回业务数据。这个add_divider就已经是一个完整的装饰器了。它接收get_user,用wrapper把它包起来,wrapper在调用原函数之前打印一条分割线。
显然,每次写get_user = add_divider(get_user)太啰嗦,Python 提供了@语法糖,上面的代码可以写成:
@add_divider def get_user(): return {"name": "王五", "level": "admin"}你心里要清楚一个事实:@add_divider这一行,在代码执行到这里时,真实发生的事情就是get_user = add_divider(get_user)。它会被重新赋值——get_user这个变量名不再指向你写的那个函数,而是指向wrapper。这一点很重要,后面排查"函数名对不上""注释找不到"这类问题的时候全靠它。
3.2 wrapper 里那两个星号是干什么的
新手看装饰器代码最容易疑问的地方就是*args和**kwargs。其实它们的意思很简单:
*args用于收集所有位置参数,打包成一个元组**kwargs用于收集所有关键字参数,打包成一个字典
之所以 wrapper 一定要写这两个东西,是因为装饰器要包裹的函数你事先不知道长什么样。它可能接收两个参数,可能接收五个,可能一个都不接收,可能全部是关键字参数。为了让装饰器具备通用性,wrapper 就用*args, **kwargs把函数调用时传进来的参数原样收下,之后再原样转传给原函数。
我曾经见过有人偷懒,wrapper 写成def wrapper(),调用原函数时也不传参。这样当天写的装饰器碰到一个接收参数的函数就会直接报TypeError。装饰器存在的意义就是复用,你包装的函数千奇百怪,参数签名五花八门,所以*args, **kwargs不是可选项,而是必备项。
3.3 functools.wraps 到底解决了什么问题
前面我提醒过,一旦加了装饰器,get_user这个名字指向的是 wrapper 而不是原函数。这带来两个副作用:函数的名字变成wrapper,函数的文档字符串变成空。
def add_divider(func): def wrapper(*args, **kwargs): """I am wrapper""" print("=" * 30) return func(*args, **kwargs) return wrapper @add_divider def get_user(): """返回当前用户信息""" return {"name": "王五", "level": "admin"} print(get_user.__name__) # wrapper print(get_user.__doc__) # I am wrapper这在调试和自动化文档生成时会带来困扰。解决办法是functools.wraps,它在装饰器内部把原函数的元信息复制到 wrapper 上,同时更新 wrapper 的__dict__,让get_user.__name__和get_user.__doc__恢复成原函数的值。
实际项目里我要求所有装饰器都必须加@wraps(func)。有一个亲身教训:某次线上排查问题,报错堆栈里面显示的调用方是wrapper,而不是业务函数名get_order。因为同名 wrapper 太多,我去代码里搜 "wrapper" 搜出来一屏,定位具体是哪个业务函数花的排查时间比平时多了好几倍。加了@wraps之后报错会显示函数真实的名字,帧一眼就能看清调用链。
4. 带参数的装饰器,三层嵌套的玄机
4.1 为什么需要三层函数
很多装饰器本身需要配置参数。比如一个缓存装饰器,你想指定缓存过期时间;一个权限校验装饰器,你想指定需要的角色。这时你会想写:
@cache(ttl=60) def get_user(): ...注意,这里的cache(ttl=60)是函数调用,它返回的才是真正的装饰器。所以带参数的装饰器天然就是三层嵌套:外层接收装饰器参数,中层接收原函数,内层 wrapper 接收业务函数调用时的参数。
import time from functools import wraps def cache(ttl=60): def decorator(func): cached_data = {} @wraps(func) def wrapper(*args, **kwargs): key = (args, tuple(sorted(kwargs.items()))) if key in cached_data: print("hit cache") return cached_data[key] result = func(*args, **kwargs) cached_data[key] = result return result return wrapper return decorator @cache(ttl=120) def get_user(user_id): time.sleep(1) return {"user_id": user_id, "name": "张三"}三层分别干什么:
cache(ttl=120)执行,返回decorator,同时ttl被闭包记住@decorator生效,相当于get_user = decorator(get_user),decorator接收原函数,返回wrapperwrapper里存了原函数,调用get_user(1001)时,wrapper先查缓存,再决定是否调用原函数
有朋友问我什么时候该用三层。我一般这样判断:装饰器名字后面要不要跟括号。@logger和@logger("api")完全是两种写法。只要装饰器本身需要用户传入参数,就得用三层结构。
4.2 functools.partial 实现带参数装饰器的小技巧
三层嵌套写多了之后会嫌它丑,有人用functools.partial来实现带参数装饰器,把接收原函数和接收参数两次调用合并处理:
from functools import partial, wraps def cache(func=None, *, ttl=60): if func is None: return partial(cache, ttl=ttl) @wraps(func) def wrapper(*args, **kwargs): ... return wrapper # 两种用法都支持 @cache def foo(): ... @cache(ttl=120) def bar(): ...这个技巧的优点是一个装饰器能同时兼容@cache和@cache(ttl=120)两种用法。缺点是对新手来说可读性差一些,团队协作时如果成员水平参差不齐,代码 review 环节容易炸。我的建议是项目里保持一种写法,要么统一三层,要么统一 partial 的写法,最怕的是两种都有。
5. 类装饰器与带状态的装饰器
5.1 用类实现装饰器
函数式写法能解决大部分需求,但有时候你要在装饰器里维护的状态比较多,比如计数器、重试次数、上次执行时间,用函数写闭包虽然也不难,但可读性和可维护性不如类。类实现装饰器依赖__call__方法,实例被调用时会触发它。
from functools import wraps class Retry: def __init__(self, times=3, delay=0.5): self.times = times self.delay = delay self.call_count = 0 def __call__(self, func): @wraps(func) def wrapper(*args, **kwargs): for i in range(self.times): self.call_count += 1 try: return func(*args, **kwargs) except Exception as e: print(f"第 {i+1} 次调用失败: {e}") time.sleep(self.delay) raise RuntimeError(f"重试 {self.times} 次仍然失败") return wrapper @Retry(times=3, delay=1) def unstable_api(): if random.random() < 0.5: raise ConnectionError("网络异常") return "success"类装饰器的好处是状态看得见。比如你要统计一个函数总共被调用多少次、失败多少次、平均耗时多少,把所有计数器作为实例属性挂在外面,别的代码也能随时访问。有一次线上系统出问题,我写了一个带重试机制的装饰器,把每次失败的原因和次数都记录在实例属性里,直接通过调试控制台就能看到某条链路在哪个环节重试了几次,排查效率比看日志强得多。
5.2 带初始化的类装饰器
上面这种写法还有一个细节值得注意:Retry(times=3, delay=1)是在装饰时才接收参数的,如果你还想"装饰器接收函数的同时接收参数",可以通过新增一个实例方法接口来区分。很多人会跟"带参数的函数式装饰器"搞混,其实类装饰器天然就把"参数"放进了__init__,把"函数"放进__call__,结构上比三层嵌套更清晰。
class Retry: class Inner: def __init__(self, retry_obj, func): self.retry_obj = retry_obj self.func = func functools.update_wrapper(self, func) def __call__(self, *args, **kwargs): for attempt in range(self.retry_obj.times): try: return self.func(*args, **kwargs) except Exception as e: ... raise ... def __init__(self, times=3): self.times = times def __call__(self, func=None, **kwargs): if func is None: return lambda f: self.Inner(self, f) return self.Inner(self, func)这种写法等于把"函数式三层嵌套"翻译成"类 + 内部类"的组合,让每个角色的职责更加明确。不过大多数业务场景真的不需要把装饰器写到这么复杂,我在项目里见过的大部分装饰器用函数式写法就足够干净,类装饰器更适合那种"装饰器本身要长期维护、内部状态需要扩展"的场景。
6. 实操中高频踩坑与排查实录
6.1 闭包变量绑定问题
经典的坑出现在循环里定义装饰器或者用装饰器生成函数。下面这个例子是网上流传很广的"陷阱题":
def make_multipliers(): multipliers = [] for i in range(3): def multiplier(x): return x * i multipliers.append(multiplier) return multipliers for m in make_multipliers(): print(m(2)) # 输出全是 4,而不是 0, 2, 4原因很简单:闭包捕获的是变量i本身,而不是i当时的值。循环结束后i的值是 2,所以所有函数取到的都是 2。这跟装饰器有什么关系?有,如果你在循环里批量给函数加装饰器,装饰器内部如果引用了循环变量,同样会踩这个坑。
解决办法是利用默认参数的绑定时机,或者用functools.partial:
def multiplier(x, i=i): return x * i把i作为默认参数后,循环每次迭代都会把当时的i值固化进函数定义里。这个坑我在实际写批量注册装饰器时踩过一次,排了半天才发现是闭包延迟绑定的事。以后凡是循环里使用循环变量构造闭包,我第一反应就是查绑定问题。
6.2 装饰器执行顺序
多个装饰器叠在一起的顺序问题,也是新手常常搞反的。记住一句话:从上往下装饰,从下往上执行。
@login_required @log_execution def sensitive_op(): ...等价于:
sensitive_op = login_required(log_execution(sensitive_op))所以当调用sensitive_op()时,先执行的是login_required的 wrapper 逻辑,再往下执行log_execution的 wrapper 逻辑,最后才到达原函数。权限校验这种"必须最早执行"的逻辑要放在最上面,日志这种"任何情况都可能要打"的逻辑放下面。
我分享一个具体经验:曾经写一个支付相关的接口,叠加了三个装饰器——@login_required、@log_execution、@validate_params。结果参数校验失败时日志却没有记录到,因为validate_params在最下面,参数校验失败直接把函数截断了,而log_execution根本没机会执行。后来把@log_execution挪到最上面,所有请求都先进日志,再进后续校验,问题才解决。装饰器的顺序直接影响逻辑层级,务必先想清楚哪一层是"最外部"。
6.3 丢失函数元信息
不加@wraps的后果前面已经说过了,这里再补充一个更隐蔽的后果:如果当时你用了inspect.signature来分析函数参数,或者用了 FastAPI 这类依赖函数签名做参数注入的框架,丢失元信息会让程序直接报错或者行为异常。之前用 FastAPI 写一个小服务,某个接口的依赖注入突然全部失效,查了很久发现是中间层有人给视图函数加了一个不带@wraps的装饰器,FastAPI 拿到的是 wrapper 的签名,自然分析不出参数。
所以一个最简单也最重要的习惯:每个装饰器的 wrapper 上都要写@wraps(func),这应该是装饰器代码的第一条纪律。
6.4 调试技巧
加了装饰器之后,pdb调试和异常堆栈都会被包裹层影响。除了@wraps之外,还有几个技巧:
- 使用
__wrapped__属性。@wraps会把原函数赋给wrapper.__wrapped__,调试时可以直接取到原始函数 - 需要临时快速查看原函数的话,可以直接
func.__wrapped__拿原始对象,绕开包装逻辑 - 在装饰器内部临时加
breakpoint(),可以先看装饰器有没有被正确触发 - 如果怀疑多个装饰器之间互相干扰,逐个注释掉,用二分法定位问题层
7. 几个真实场景的完整代码方案
7.1 缓存装饰器
实际项目中,我用的缓存装饰器比前面那个示例要严肃得多。要考虑线程安全问题、缓存 key 序列化的完整性问题、ttl 过期时间的处理。一个基础可用的版本:
import time import threading from functools import wraps class TTLCache: def __init__(self, ttl=60, maxsize=128): self.ttl = ttl self.maxsize = maxsize self._data = {} self._lock = threading.Lock() def __call__(self, func): @wraps(func) def wrapper(*args, **kwargs): key = self._make_key(args, kwargs) now = time.time() with self._lock: if key in self._data: value, expire_at = self._data[key] if expire_at > now: return value del self._data[key] result = func(*args, **kwargs) self._data[key] = (result, now + self.ttl) if len(self._data) > self.maxsize: self._evict() return result return wrapper def _make_key(self, args, kwargs): # 实际项目中建议用序列化或 repr 保证不同参数生成不同 key return (repr(args), repr(sorted(kwargs.items()))) def _evict(self): # 简单的淘汰策略:删除最早写入的一项 oldest_key = next(iter(self._data)) del self._data[oldest_key] @TTLCache(ttl=30) def get_stock_price(code): # 模拟从接口拉取股票价格 time.sleep(1) return {"code": code, "price": 12.34}注意_make_key里我用了repr,实际生产环境应该用更可靠的序列化策略,比如把args里各种类型统一转换为可哈希的字符串。有人用hash(args)直接当 key,字典的hash有随机化问题,进程重启之后就失效了,不可取。
7.2 重试装饰器
重试装饰器是典型的"切面"逻辑。我把它单独抽出来成一个公共组件,所有调第三方接口、数据库连接池获取、RPC 调用的地方都能复用:
import time import logging from functools import wraps logger = logging.getLogger(__name__) def retry(max_attempts=3, delay=0.5, backoff=2, exceptions=(Exception,)): """ max_attempts: 最大尝试次数 delay: 初始重试延迟 backoff: 每次重试延迟倍数 exceptions: 捕获的异常类型,默认捕获所有异常 """ def decorator(func): @wraps(func) def wrapper(*args, **kwargs): current_delay = delay for attempt in range(1, max_attempts + 1): try: return func(*args, **kwargs) except exceptions as e: if attempt == max_attempts: raise logger.warning(f"{func.__name__} 第 {attempt} 次调用失败:{e}, {current_delay} 秒后重试") time.sleep(current_delay) current_delay *= backoff return wrapper return decorator这个版本的亮点是backoff指数退避,避免重试时把下游系统打崩。看过太多"重试三次,间隔 0 秒"的代码,下游本来就因为压力大而超时,你还猛打三次,不雪崩才怪。合理的退避策略是:第一次失败等 0.5 秒,第二次等 1 秒,第三次等 2 秒。这个技巧算是重试组件的标配了。
7.3 权限校验装饰器
权限校验装饰器最大的价值在于把"谁有权限"这类规则,从业务函数里剥离出来。可以设计成既支持直接传角色,也支持从函数内部自定义校验函数:
from functools import wraps def require_permission(permission: str = None): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): user = getattr(args[0], "user", None) if args else None if not user: raise PermissionError("未登录") if permission and permission not in user.permissions: raise PermissionError(f"缺少权限: {permission}") return func(*args, **kwargs) return wrapper return decorator实际项目里权限校验往往需要从请求上下文里取用户信息,所以这个装饰器通常不是通用的。我更推荐把权限规则放在装饰器参数里,由装饰器内部统一从全局 context 或者请求对象中获取用户,这样一个项目里所有接口的权限判断逻辑都是同一套实现,审计的时候也方便。
7.4 性能统计装饰器
这个装饰器是排查性能问题时我第一个会挂上的东西,也是帮你快速确认"慢在哪"的有效工具:
import time from functools import wraps def elapsed_time(func): @wraps(func) def wrapper(*args, **kwargs): start = time.perf_counter() result = func(*args, **kwargs) elapsed = time.perf_counter() - start print(f"{func.__name__} 耗时 {elapsed * 1000:.2f} ms") return result return wrapper我常常在定位慢接口时临时给可疑函数挂上这个装饰器,跑一遍就能看出耗时大头。注意用time.perf_counter而不是time.time,perf_counter是单调递增的,不会因为系统时间被改而受影响,测量短间隔更精确。
7.5 日志装饰器
最后推荐一个务实版的日志装饰器,自带异常堆栈信息,出问题后人不用翻冗长的日志去拼调用链:
import logging import traceback from functools import wraps logger = logging.getLogger("service") def log_call(func): @wraps(func) def wrapper(*args, **kwargs): logger.info(f"开始调用 {func.__name__}, args={args}, kwargs={kwargs}") try: result = func(*args, **kwargs) logger.info(f"调用完成 {func.__name__}, result={result}") return result except Exception: logger.error(f"调用失败 {func.__name__}, 堆栈:\n{traceback.format_exc()}") raise return wrapper这个装饰器能直接解决"线上出了问题不知道数据流怎么走"的痛点。我习惯在调试阶段给关键链路的每个环节加上,确认没问题后再删掉部分日志,避免生产环境日志噪音过大。
8. 常见问题速查表
整理一个列表,方便你们收藏后随时查阅:
| 现象 | 原因 | 解决办法 |
|---|---|---|
函数名变成wrapper | 没有使用@wraps | 装饰器内from functools import wraps,并在 wrapper 上标注 |
| 函数文档字符串丢失 | 没有使用@wraps | 同上 |
| FastAPI/Flask 路由参数解析失败 | 装饰器覆盖了原函数签名 | 加@wraps,某些框架还需要inspect.signature方法保留参数注解 |
| 循环中定义的函数全部取到最后一次循环值 | 闭包延迟绑定 | 默认参数绑定i=i或使用functools.partial |
| 多个装饰器顺序导致权限校验绕过日志 | 装饰器顺序写反 | 按"最外层最先执行"记忆:@login_required @log_execution时 login 先执行 |
| 重试时下游反而被压垮 | 没有退避策略,重试间隔太短 | 加指数退避delay *= backoff |
| 装饰器内修改了可变对象,函数间互相影响 | 闭包共享状态 | 每次调用前初始化状态,或在 decorator 内部新建状态容器 |
| 带参数装饰器和无参数装饰器语法混淆 | 三层和两层的结构没分清 | 无参数用两层;有参数用三层或functools.partial兼容两种 |
| 性能统计不准 | 用了time.time() | 改用time.perf_counter() |
这个表格里每一条都是我或者身边的人真实踩过的,尤其是"装饰器顺序"和"闭包绑定"这两个,是面试题里高频出现、日常编码里同样高频翻车的存在。
9. 我平时用得顺手的一些经验
最后分享几个可能帮到你的习惯。
第一个习惯:装饰器文件单独放。我在项目里喜欢建一个decorators.py模块,把定时重试、缓存、日志、权限校验这类通用装饰器全部收拢在一起,加好单元测试。这样业务代码里 import 一下就能用,也意味着所有项目统一通过这些公共组件做切面控制。出了问题时,只需要查一个文件,不用翻遍整个仓库找装饰器定义。
第二个习惯:装饰器本身也要写异常处理。我看到很多装饰器只关注"装饰"这个动作,wrapper 内部没有try/except,结果原函数抛异常时,装饰器里面的日志也打不出来,排查时陷入混乱。但凡做日志或统计的装饰器,我强烈建议在 wrapper 里面把异常记录一下再重新抛出。
第三个习惯:给装饰器写单元测试。装饰器是公共逻辑,一旦错了影响面很大。测试不太好写,但至少要覆盖以下几条:原函数正常调用时行为正确、原函数抛出异常时行为正确、带参数与不带参数的调用方式都兼容、函数元信息被正确保留。
第四个习惯:学会用__wrapped__绕过装饰器。做测试或者调试时,如果只想调用原始函数而不想触发装饰器的副作用,直接:
func = get_user.__wrapped__ func(1001)这个技巧在离线脚本、数据修复任务里特别有用。有一次我们需要批量重算一批旧数据,但业务函数上挂了一个缓存装饰器,直接调会返回旧缓存导致数据不对。用__wrapped__绕过缓存之后问题立刻解决。
讲讲我自己的经验吧。早期写项目时,我总爱把装饰器用在各种想得到的地方,觉得它很炫酷。后来维护老代码时发现,装饰器用多了以后,代码的可读性和可追踪性会下降。一个函数被五六个装饰器包着,读代码时得一层层剥开,再加上没有@wraps,连函数名都变了,排查一次问题非常痛苦。现在的原则是:能用装饰器逻辑收敛的重复代码就用,但不要为了用而用。清洗共享逻辑与保持代码可读之间,需要找到均衡点。
装饰器不是一个需要专门花两周去啃的知识点,它就是 Python 里一个"函数是用来被组合使用"的体现。你把它当成工具去解决实际问题,而不是当成语法表演去堆砌,理解起来会舒服得多。如果你手头正好有一个"每个函数都要做一遍"的活,比如打日志、加缓存、做权限判断,试着抽一个装饰器出来,亲手写一遍,比看十篇教程都管用。