1. 先从一段"看起来很魔法"的代码说起
刚接触Python的人,十有八九见过这样的代码:
@timer def get_data(): ...然后心里冒出一个问号:这个带@的玩意儿到底是什么?为什么它能把一个普通函数"包一层"?
我当初学装饰器的时候,也经历过一段"会用但说不清"的迷糊期——知道拿@login_required往视图函数上一放就能做登录校验,知道@staticmethod可以让方法变成静态方法,但你要是让我亲手写一个带参数的装饰器,我大概率会卡壳。
后来写了两年多业务代码,把日志、缓存、重试、权限校验这些逻辑反复用装饰器落地之后,我才真正摸清它的脾气。这篇就好好聊聊Python装饰器到底怎么回事、怎么写不踩坑、以及实际项目里它到底是用来解决什么问题的。不管你是刚写完爬虫、正准备做数据分析的初学者,还是已经写了一些业务代码想进阶的开发者,这篇文章都会对你有用。
我要先强调的是:装饰器不是一个"高级到只有大神才能用"的东西,它本质上就是一个普通函数,只不过这个函数的参数和返回值都是函数。搞懂这一点,后面所有花哨写法都不算事。
2. 装饰器到底在做什么——先拆开揉碎看本质
2.1 函数在Python里也是"数据"
要说清楚装饰器,必须先接受一个对很多初学者来说有点反直觉的事实:在Python里,函数是一种对象,可以像整数、字符串一样被传递、被返回、被赋值给变量。
def say_hello(): return "hello" # 把函数赋值给另一个变量 greet = say_hello print(greet()) # 输出 hello # 把函数当作参数传进另一个函数 def call(func): return func() print(call(say_hello)) # 输出 hello这里你看到的两件"理所当然"的事——函数能赋值、函数能传参——就是装饰器的地基。用生活化的话来说:函数就像一把通用的工具,你可以把这把工具递给别人用,别人也可以把这把工具加工一番再还给你。
2.2 装饰器的原始形态:就是一个"包装函数"
既然函数能接收函数作为参数,也能返回一个新的函数,那我不就可以写这样一个东西:
def my_decorator(func): def wrapper(): print("调用前,做点什么") result = func() print("调用后,做点什么") return result return wrapper这个my_decorator接收一个函数func,然后在内部定义一个wrapper,在调用func前后各塞了一段逻辑,最后把这个wrapper返回出去。
注意:到这一步,原来的func并没有被修改。Python里的函数对象一旦创建,它的代码体是固定的,装饰器做的事是"拿一个新函数把它包起来",调用的时候实际上跑的是新函数,新函数内部再调旧函数。
这种"在一个函数外面再包一层函数"的做法,跟快递柜的包裹代收很像:你原本的收件流程是"快递员→你",加了快递柜之后变成"快递员→快递柜→你"。快递柜可以在中间帮你完成代收、保管、记录,你拿到包裹的体验其实没变,但整个过程多了一层可控性。
2.3@语法糖到底做了什么
上面那种写法在Python 2时代就有了,但写起来很啰嗦:
def get_data(): ... get_data = my_decorator(get_data)每次装饰完还要重新赋值一次,非常容易漏写。Python 2.4引入了@语法糖,让你可以直接这样写:
@my_decorator def get_data(): ...这两段代码在语义上完全等价。@my_decorator放在函数定义上方,Python解释器执行到这一行的时候,会把下面定义的函数对象取出来,传给my_decorator,再把返回值重新赋值给函数名get_data。
所以@不是玄学,它只是一个"我懒得写get_data = my_decorator(get_data)"的快捷方式。语法糖的本质是简化写法,不改变底层机制。
注意:装饰器是在函数定义的时候立刻执行的,不是在函数调用的时候执行的。很多新手以为
@是在函数每次被调用时才"包装"一次,其实不对。模块加载时,装饰器就跑完了,wrapper已经准备好了,后面每次调用都是直接调wrapper。
2.4 为什么需要装饰器——从实际问题出发
假设你在写一个电商系统的下单接口,随着需求迭代,你先后需要给这个接口加这些功能:
- 记录调用日志
- 统计耗时
- 校验用户是否登录
- 万一调用失败要自动重试
- 把计算结果缓存起来
如果直接在业务函数里加这些逻辑,函数会迅速膨胀到几百行,而且下单接口、查询接口、退款接口都要面对相同的需求,你会在每个函数里复制粘贴同一段逻辑。
装饰器的价值,就是把横切在多个函数头上的这些"通用旁路逻辑"抽离出来,写成独立的、可复用的组件,然后像贴标签一样往函数上一放。
这跟你装修房子是一样的思路:水管电路(日志、鉴权、缓存)是横贯整个房子的公共设施,不是只在卧室里才有需求。把它们集中布好,每个房间想用就直接接上,而不是每个房间单独埋一套。
3. 核心细节与实操要点——写出不炸的装饰器
3.1 无参装饰器:最基础的写法
先看一个最简单的无参装饰器,它不接收额外参数,只包装目标函数:
import functools import time def timer(func): @functools.wraps(func) def wrapper(*args, **kwargs): start = time.perf_counter() result = func(*args, **kwargs) cost = time.perf_counter() - start print(f"函数 {func.__name__} 耗时 {cost:.6f} 秒") return result return wrapper @timer def compute(): time.sleep(0.1) return 42 print(compute())这里的*args, **kwargs是装饰器里的关键点。你写的普通函数可能有位置参数、关键字参数、可变参数,为了让wrapper能转发任意形式的参数,最稳妥的写法就是def wrapper(*args, **kwargs): ...,然后在内部用func(*args, **kwargs)调用原函数。这样它就能适配任何签名。
为什么需要functools.wraps?后面[章节4]我会详细讲,这里先记住一行字:@functools.wraps(func)能把原函数的名字、文档字符串、参数信息复制到wrapper上,否则你的函数在别人眼里会变成一个叫wrapper的陌生人。
3.2 带参装饰器:多套一层"工厂"
很多场景下,装饰器本身需要参数。比如你想控制日志级别:
import functools import logging def log_with(level): def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): logging.log(level, f"调用 {func.__name__}") return func(*args, **kwargs) return wrapper return decorator @log_with(logging.INFO) def business_func(): ...这个写法很多人第一次看会懵:为什么外面还要包一层?
因为@log_with(logging.INFO)的执行顺序是:先调用log_with(logging.INFO),得到decorator这个函数,再用它去装饰business_func。也就是说,你希望"日志级别"这个参数在装饰之前就确定下来,所以必须通过外层函数的参数传递。
用一句话记住结构:
- 最外层(
log_with):接收装饰器的参数,返回内层装饰器 - 中间层(
decorator):接收目标函数,返回wrapper - 最内层(
wrapper):接收原函数调用时的参数,执行真正的逻辑
带参装饰器相当于一个"装饰器工厂"。淘宝上买的普通贴纸(无参装饰器)是现成的,直接贴就行;定制贴纸(带参装饰器)需要你先告诉店家印什么字,等他做出来你再贴。这个"先定制、再贴"的过程就是多包了最外层那层壳。
3.3 类装饰器:用类当装饰器
函数装饰器用久了你会发现,有的场景需要保存状态,比如统计一个函数被调用了多少次。用函数闭包写也行,但用类会更直观,因为类的实例天然可以挂在状态:
import functools class Counter: def __init__(self, func): functools.update_wrapper(self, func) self.func = func self.count = 0 def __call__(self, *args, **kwargs): self.count += 1 print(f"调用次数: {self.count}") return self.func(*args, **kwargs) @Counter def ping(): return "pong" ping() ping() print(ping.count) # 2这里@Counter等价于ping = Counter(ping)。由于Counter实例实现了__call__方法,它本身可调用,所以ping()最终会走到实例的__call__,在调用原函数之前把计数加一。
类装饰器适合什么场景?
- 需要维护装饰器自身的状态(计数器、缓存容器、连接池)
- 想把逻辑组织得更清晰,方便扩展方法
但它也有缺点:类实例在内存里的开销比闭包大一点,而且如果你不调用functools.update_wrapper,函数的元信息同样会丢失。所以我个人建议:简单的用函数装饰器,复杂的才上类装饰器。
3.4 内置装饰器:你其实早就用过了
Python自带了不少装饰器,很多人在写代码时天天用却没意识到它们是装饰器:
@property:把一个方法变成属性访问,比如obj.name而不是obj.name()@staticmethod:静态方法,不需要实例也能调用@classmethod:类方法,接收cls作为第一个参数@functools.lru_cache:对函数结果做缓存,递归计算时效果极佳@dataclass(严格来说是类装饰器):自动生成__init__、__repr__等方法
以lru_cache为例,它解决的是"同一个输入反复计算"的问题:
import functools @functools.lru_cache(maxsize=128) def fib(n): if n < 2: return n return fib(n - 1) + fib(n - 2)lru_cache内部维护了一个字典,把(n)映射到计算结果。下次调用如果参数相同,直接从缓存里取,不再执行函数体。用递归算斐波那契数列时,这个装饰器能把时间复杂度从指数级降到线性级,效果立竿见影。
3.5 多个装饰器叠加时的执行顺序
装饰器可以叠加,但顺序不能乱:
@a @b def f(): ...它等价于f = a(b(f))。什么意思?
- 先执行
b(f),得到b_f - 再执行
a(b_f),得到a_b_f - 最终
f指向a_b_f - 每次调用
f(),先进入a的逻辑,再进入b的逻辑,最后才执行原来的函数体
这就是"洋葱模型"。你从最外层切进去,逐层剥到最里面,再逐层退出来。
所以在实际业务里,装饰器的顺序是有讲究的。比如一个视图函数需要同时做登录校验和权限校验,如果@login_required写在@permission_required外面,那么请求会先经过登录校验,再经过权限校验。如果你顺序写反了,可能出现"未登录用户先被拿去验证权限"这种尴尬局面。
规则很简单:先执行的装饰器(离函数最近的那个)先包装,后执行的装饰器在最外层,调用时最外层的逻辑先跑。
4. 实操过程与核心环节实现——四个可落地的装饰器
下面我用一个接近真实的业务场景,把装饰器完整的落地过程走一遍。假设你正在写一个订单处理的脚本,涉及从外部API拉取数据、解析、写库的完整流程,我要在不污染业务函数的前提下,给它加上日志、耗时统计、失败重试和结果缓存四个能力。
4.1 第一步:搭建基础函数
先写一个模拟的业务函数,假装从某个接口拉取订单数据:
import random def fetch_orders(): """从订单服务拉取当天的订单列表""" # 模拟网络请求,偶尔失败 if random.random() < 0.3: raise ConnectionError("网络超时,连接订单服务失败") return [{"order_id": i, "amount": i * 100} for i in range(3)]现在这个函数有两个问题:
- 它有30%概率抛异常,直接调用的话程序就断了
- 每次调用都重新"请求"一次,没有缓存,浪费资源
4.2 第二步:写日志装饰器
日志装饰器的核心需求:记录函数名、参数、返回值或异常,但不能改变原函数的返回结果。
import functools import logging import time logging.basicConfig(level=logging.INFO) def log_call(func): @functools.wraps(func) def wrapper(*args, **kwargs): logging.info(f"开始调用 {func.__name__}, args={args}, kwargs={kwargs}") try: result = func(*args, **kwargs) logging.info(f"{func.__name__} 执行成功, 结果长度: {len(result) if hasattr(result, '__len__') else 'N/A'}") return result except Exception as e: logging.error(f"{func.__name__} 执行失败: {e}") raise return wrapper注意这里的try/except结构:不要只在失败时打日志然后吞掉异常,正确的做法是记录完日志后raise把异常继续抛出去,让上层调用方决定怎么处理。日志装饰器只负责"记录",不负责"处理"。
注意:日志装饰器里
len(result)不能直接写,因为有些函数返回的是整数、布尔值,不一定有长度。我用了hasattr(result, '__len__')先判断一下,这是一个很实际的防御性写法。
4.3 第三步:写耗时统计装饰器
耗时统计的关键点是起点和终点的采样时机。起点应该在func调用之前,终点在func返回之后。如果你把采样点放在wrapper的外面,得到的时间会包含所有内部逻辑的耗时,这在多个装饰器叠加时会产生偏差。
def time_it(func): @functools.wraps(func) def wrapper(*args, **kwargs): start = time.perf_counter() result = func(*args, **kwargs) cost_ms = (time.perf_counter() - start) * 1000 logging.info(f"{func.__name__} 耗时 {cost_ms:.2f} ms") return result return wrapper这里用time.perf_counter()而不是time.time(),是因为perf_counter是专门为测量短时间间隔设计的,精度更高,不受系统时间调整的影响。统计某段代码的耗时,perf_counter是首选。
耗时统计装饰器和日志装饰器叠加使用时,执行顺序会影响计时的范围。比如:
@log_call @time_it def fetch_orders(): ...调用时,先进入log_call的wrapper,记录"开始调用",然后调用time_it的wrapper,在这里计时,然后调用原函数。计时范围只包含原函数本身,不包含日志装饰器自己的打印耗时。如果你把顺序倒过来:
@time_it @log_call def fetch_orders(): ...计时的起点在log_call的入口,终点在它的出口,打印日志的时间也被算进去了。这个差别通常只有零点几毫秒,但在精确测量的场景下会有影响,需要心里有数。
4.4 第四步:写失败重试装饰器
重试装饰器是业务里最常用的。它的核心参数是retries(重试次数)和delay(每次重试的间隔时间)。
import time def retry(retries=3, delay=1, exceptions=(Exception,)): def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): for attempt in range(1, retries + 1): try: return func(*args, **kwargs) except exceptions as e: if attempt == retries: logging.error(f"{func.__name__} 重试 {retries} 次后仍失败: {e}") raise logging.warning(f"{func.__name__} 第 {attempt} 次调用失败: {e}, {delay} 秒后重试") time.sleep(delay) return wrapper return decorator有几个细节是真正踩过坑才会注意到的:
exceptions=(Exception,)这个参数很重要。如果你不指定,默认捕获所有Exception,包括KeyboardInterrupt和系统退出信号吗?不会,BaseException才包含那些,Exception不包括KeyboardInterrupt。但你也要注意:有些异常不值得重试,比如参数错误、权限不足,这些重试一百次也不会成功。所以调用方最好显式指定重试哪些异常,例如@retry(retries=2, exceptions=(ConnectionError, TimeoutError))。return func(*args, **kwargs)在try块里,一旦成功就立刻返回,不会继续进入下一次循环。你不需要额外写break或else。- 重试前的
delay要不要加抖动?在高并发场景下,所有失败请求同时睡一样的秒数再同时重试,会造成"惊群效应"。如果你用在爬虫或定时任务里,建议delay = delay * random.uniform(0.5, 1.5)加一点随机抖动。
4.5 第五步:写结果缓存装饰器
数据重复计算是性能浪费的大头。缓存装饰器的核心是把函数参数和计算结果放进一个字典里,下次同样的参数直接命中返回。
def cache_result(func): cache = {} @functools.wraps(func) def wrapper(*args, **kwargs): # 构建缓存key,注意顺序无关的kwargs要排序 key = (args, tuple(sorted(kwargs.items()))) if key in cache: logging.info(f"{func.__name__} 命中缓存, 返回上次结果") return cache[key] result = func(*args, **kwargs) cache[key] = result return result return wrapper手动写缓存装饰器要特别注意一点:key必须能被哈希,否则字典会报错。args和tuple(sorted(kwargs.items()))都是可哈希的元组,可以放心用。但如果args里的元素是列表、字典这类可变对象,就需要先转成元组或做序列化。
不过说实话,生产环境里一般不要自己写缓存装饰器,直接用functools.lru_cache省心很多。我写这个是为了展示内部原理,让你知道lru_cache不是魔法,它就是一个在函数外面包了层带字典缓存的wrapper。
4.6 完整组合:四个装饰器一起上
把前面几个装饰器组合在一起,看一下完整效果:
@log_call @time_it @retry(retries=3, delay=0.5, exceptions=(ConnectionError,)) @cache_result def fetch_orders(): """从订单服务拉取当天的订单列表""" if random.random() < 0.3: raise ConnectionError("网络超时,连接订单服务失败") return [{"order_id": i, "amount": i * 100} for i in range(3)] for _ in range(5): try: print(fetch_orders()) print("---") except Exception: print("最终失败") break这个叠加很有意思。按洋葱模型分析:
@log_call是最外层:记录开始、成功或失败的日志@time_it在第二层:统计整个调用的耗时@retry在第三层:捕获ConnectionError并重试@cache_result在最内层:先查缓存,如果命中了立即返回,不再进入fetch_orders原函数
调用流程是:log_call入口 →time_it计时开始 →retry尝试调用 →cache_result查缓存 → 如果缓存没命中,执行原函数;如果失败抛出ConnectionError,retry捕获并重试(重新走cache_result);最终成功时,按原路返回,time_it计算耗时并记录,log_call记录成功。
这里有个很隐蔽但值得注意的地方:retry里重新尝试的是"调用装饰后的函数"还是"调用原函数"?如果我把retry写在cache_result外面,那么重试时也会先查缓存,但因为第一次调用已经抛异常了,缓存里不会有记录,所以查了也白查,逻辑上没毛病。但如果把retry写在cache_result里面,每次重试的是原函数,就不能在重试之间共享缓存状态了。所以:装饰器的排列顺序,决定了每个装饰器能看到其他装饰器的行为效果。
实际建议:不要在一开始就把四个装饰器全堆上去。先加
retry解决函数不稳定,再加cache_result减少重复计算,最后按需加日志和耗时统计。每加一个就跑一遍测试用例,确保没有破坏原来的行为。
5. 常见问题与排查技巧实录
这个章节里的内容,每条都是我或身边同事实际踩过的坑,按出现频率从高到低排列。
5.1 函数名字变了:__name__被覆盖
不加functools.wraps时,原函数的所有元信息都会丢失:
def my_wrapper(func): def inner(*args, **kwargs): return func(*args, **kwargs) return inner @my_wrapper def add(): """两数相加""" ... print(add.__name__) # inner,而不是 add print(add.__doc__) # None后果是什么?
- 调试时日志里打印的函数名全部变成
inner,排查问题很痛苦 - 某些框架(比如Flask、Django)依赖视图函数的
__name__做路由注册,名字变了会导致路由功能异常 - 基于文档字符串的自动化测试工具识别不到函数
解决方案只有一个:在inner前面加@functools.wraps(func)。它做的事情是调用functools.update_wrapper,把func的__name__、__doc__、__module__、__qualname__等属性复制到inner上。
5.2 带参装饰器写成了两层而不是三层
我见过很多新手写带参装饰器时少包一层:
def log_with(level): def decorator(func): ... return decorator少了一层,把level参数和func参数放在同一个函数里:
def log_with(level, func): ... return wrapper这种写法在@log_with(logging.INFO)时直接报错,因为log_with(logging.INFO)只传了一个参数level,但func没传。用一句话记结构:"带参装饰器一定要有三层:外层接参数、中层接函数、内层接调用参数。"如果哪天你写完发现装饰器在定义阶段就报missing 1 required positional argument,九成是这里出了问题。
5.3 装饰器里的变量是共享的
看这个缓存装饰器的经典坑:
# 每次装饰都会创建一个新的cache dict吗? def cache_result(func): cache = {} ...答案是:会的。每次应用装饰器时cache_result(func)都会执行一次,函数体内的cache = {}会创建独立的新字典,互不共享。所以不同函数的缓存是隔离的。
但如果你把缓存字典定义在模块级别,那么所有被这个装饰器包装的函数会共享同一个字典,这时如果两个函数刚好有相同参数,会互相串结果:
cache = {} def cache_result(func): def wrapper(*args, **kwargs): key = (func.__name__, args, tuple(sorted(kwargs.items()))) ...所以共享状态的规则是:定义在装饰器函数内部的变量,每个被装饰函数独立;定义在外部的,大家共享。你在设计时需要想清楚要哪种。lru_cache之所以允许maxsize参数,就是让每个函数有自己独立的缓存空间。
5.4 装饰器自己吃掉了返回值或者异常
这是最隐蔽的坑。看下面的错误示范:
def log_call(func): def wrapper(*args, **kwargs): result = func(*args, **kwargs) print("调用了") # 忘记 return result return wrapper如果wrapper里面不return result,那么调用add(1, 2)会返回None,而不是3。这个问题不仅在装饰器里存在,在任何"包装函数"里都存在。你包装的那层只是中间商,不能把货吞了。
类似的,如果写:
def safe_call(func): def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception: print("出错了") # 不重新raise return wrapper异常确实被拦住了,但调用方完全不知道发生了什么,后续程序会带病运行,排查问题时你会发现一些数据写了一半、资源没释放的诡异现象。
我的经验法则是:
- 装饰器默认要返回原函数的结果
- 装饰器默认要把捕获的异常重新抛出去
- 除非你非常明确要在某一层做短路处理,否则不要轻易吞掉异常
5.5 装饰器装饰类方法时,self跑哪儿去了
装饰器用在类方法上时,有一个特殊之处:self被传入wrapper的第一个位置参数。
class Service: @log_call def handle(self, data): return data s = Service() s.handle("x") # wrapper接到的args = (s, "x")如果你的装饰器里写func(*args, **kwargs),self会被正常传递给原方法,没问题。但如果你的装饰器想额外处理"第一个参数",就要注意args[0]是self,不是业务参数。有些装饰器是给函数写的,直接拿args[0]当业务参数处理,装饰在类方法上就全乱套了。
解决方法:在装饰器内部判断原函数是不是方法。一个简单但不优雅的办法是不管,只把args和kwargs原样转发,不要自作主张拆参数。如果确实需要读取参数做决策,可以用inspect.signature来判断第一个参数名是不是self或cls。
5.6@staticmethod和@classmethod的叠加顺序
你字面意思上可以叠加:
class A: @staticmethod @my_decorator def method(): ...但叠加顺序有讲究。如果@my_decorator写在@staticmethod的外面,那么my_decorator装饰的是一个普通的函数(因为staticmethod返回的是一个描述符对象,不是普通函数),这会导致my_decorator拿到的func不是可调用的函数对象,调用时可能会报错。
正确的做法是:@staticmethod放在最外层,让@my_decorator去装饰一个普通函数,再把结果转换成静态方法:
class A: @staticmethod @my_decorator def method(): ...不过说真的,实际业务里更推荐在类方法上装饰时直接使用函数装饰器,不要搞复杂的叠加,容易给自己挖坑。等装饰器本身摸熟了再玩花样也来得及。
5.7 性能损耗问题
装饰器毕竟多包了一层函数调用,每次调用都会多几次嵌套函数执行。不过这个损耗通常极小——一次函数调用大约几微秒级别的开销。但如果你在一个大循环里调用百万次,累积损耗就明显了。
我做过一次简单的压测:一个空函数直接调用、加一层装饰器、加四层装饰器,耗时差距大约是每百万次调用差0.3到1秒。对于业务系统来说,这个差异通常可以忽略。但如果你写的是高频调用的基础库,可以考虑:
- 用
functools.lru_cache减少重复计算,牺牲一部分内存换时间 - 把装饰器写成类并用
__slots__减少实例内存 - 甚至在上线前把装饰器逻辑手工内联到关键路径上(不推荐,除非Profile数据明确指向这里)
6. 实用场景清单:从爬虫到数据分析,装饰器无处不在
聊完了写法和坑,我把实际开发中装饰器最常用的场景整理成了一张表,方便你按图索骥。我看到热搜词里有爬虫、数据分析、量化交易之类的关键词,这些领域的开发者其实每天都在和装饰器打交道,只是一些框架帮你封装好了。
| 应用场景 | 核心需求 | 推荐方案 | 注意事项 |
|---|---|---|---|
| 接口鉴权 | 校验用户身份、权限 | @login_required | 装饰器顺序要保证鉴权在外层 |
| 请求重试 | 网络不稳定时自动重试 | 自定义@retry | 指定可重试的异常类型,加随机抖动 |
| 结果缓存 | 避免重复计算 | @functools.lru_cache或@cache_result | 注意缓存键的可哈希性、缓存上限 |
| 日志追踪 | 记录调用链、参数、返回值 | 自定义@log_call | 不要吞异常,注意日志敏感信息脱敏 |
| 耗时统计 | 性能分析、慢请求定位 | 自定义@time_it | 用perf_counter而不是time.time |
| 输入校验 | 检查参数合法性 | 自定义校验装饰器 | 利用inspect.signature做参数名映射 |
| 限流/防抖 | 控制调用频率 | 自定义@rate_limit | 复杂度较高,先想清楚分布式还是单机 |
| 发布订阅 | 事件触发后自动执行 | 配合注册表实现 | 注意引用循环和内存泄漏 |
举个例子,写爬虫时最常见的需求是对每一个页面抓取请求做"重试+限速":
@retry(retries=3, delay=2, exceptions=(requests.exceptions.ConnectionError,)) @rate_limit(calls_per_second=2) def fetch_page(url): resp = requests.get(url, timeout=10) resp.raise_for_status() return resp.text这两个装饰器叠加在一起,爬虫脚本的健壮性立刻提升一个档次。同样的思路适用于量化策略中回测数据的拉取、交易接口的调用,它们的共同特点是:外部服务不稳定,但你不能因为这个不稳定就让整个程序崩溃。
数据分析场景一个很实用的用法是:把 DataFrame 的计算过程用缓存装饰器包起来,避免每次重启 Jupyter Notebook 都要重复跑一遍重型预处理。
@cache_result def load_and_clean_data(): df = pd.read_csv("raw.csv") df = df.dropna().query("amount > 0") return df加上这个装饰器后,同一个会话里多次调用load_and_clean_data(),第二次开始直接返回缓存结果,省去每次十几秒的读表清洗时间。
还有一个是很多框架帮你内置好的:Flask的路由注册本身就是一个装饰器模式、Django的@login_required、pytest的@pytest.fixture、celery的@app.task。你学会了自己写装饰器之后,再去看这些框架源码,会发现它们底层都是用同样的原理实现的——一个接收函数、返回新函数的函数。
7. 再进一步:类装饰器和上下文管理器的配合
最后聊一个稍微进阶一点但也非常实用的点:装饰器和上下文管理器(with语句)怎么配合。
有时候你希望函数在执行时临时切换状态,比如修改全局配置、临时加锁、临时换数据库连接。这种"进入时设置、退出时恢复"的场景,用上下文管理器很顺手。但如果你希望这个行为能套用在多个函数上,装饰器就是更好的载体。
import contextlib import threading def locked(lock): def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): with lock: return func(*args, **kwargs) return wrapper return decorator lock = threading.Lock() @locked(lock) def update_account(): ...with lock:把锁的获取和释放都管理好了,装饰器只需要确保函数调用在with块内即可。这种方式比在每个函数里手写lock.acquire()和lock.release()干净得多,也不容易因为异常漏掉release。
同样的思路还可以用于:
- 临时修改环境变量
- 临时切换当前工作目录
- 数据库事务提交/回滚
- 修改全局日志级别后恢复原级别
你在装饰器里用with语句,实际上就是在"包装函数"的wrapper里嵌套一个上下文管理器,让它的__enter__和__exit__自动包住原函数的执行过程。这比手动在try/finally里写恢复逻辑要优雅得多,也隐藏了异常安全细节。
我个人最喜欢的用法是"数据库事务装饰器":
def transactional(session): def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): try: result = func(*args, **kwargs) session.commit() return result except Exception: session.rollback() raise return wrapper return decorator这个装饰器保证:函数里只要抛了任何异常,事务一定回滚;只要正常返回,一定提交。你写业务函数的时候完全不用操心事务边界,只需要在函数上贴@transactional(session)这个标签,提交和回滚的脏活全交给装饰器干。
这就是装饰器最让我着迷的地方:它把"横切关注点"从业务逻辑里剥离出来,让代码的关注点单一化。当你跟别人协作时,你可以说"你只需要专注写清楚业务规则,通用的约束和增强我来管",这种代码组织方式的收益会随着项目规模扩大变得越来越明显。
如果让我给一个学习路径的建议,我会说:
- 先在简单脚本里把
wrapper(*args, **kwargs)和functools.wraps用熟 - 然后尝试写一个带参装饰器,理解三层嵌套的动机
- 再尝试写一个类装饰器,感受状态维护的不同方式
- 最后回到你正在用的框架里,挑一个内置装饰器读一读源码(
lru_cache和Flask.route源码都值得一读),你会发现之前觉得"框架帮我们搞定的事"不再是黑盒
踩过几次坑之后,我最大的体会是:装饰器不是越高级越复杂越好,而是越透明越好。好的装饰器不会改变原函数的行为契约,只是在外面加了一层可控的外壳。写的时候多想一步"如果我装饰的函数被调用一万次、被放在一个很大的代码库里、被其他同事阅读,它会不会造成困惑",很多设计问题就不会发生。