Python装饰器
Python 装饰器:别被 @ 符号吓到,它就是个”包装盒”
第一次看到 @ 符号的时候,我也觉得像魔法——在函数上面加一行,行为就变了,完全不知道发生了什么。
后来明白了:装饰器就是个”包装盒”。你把一个函数放进去,它给你包装一下,加些功能,再还给你。函数还是那个函数,只是多了点本事。
装饰器到底是什么?三个例子看懂
例子1:函数可以赋值给变量
def greet(name):
return f"Hello, {name}"
say_hello = greet # 把函数赋值给变量
say_hello("Alice") # "Hello, Alice"
函数跟数字、字符串一样,可以到处传递。
例子2:函数可以作为参数传给另一个函数
def apply(func, value):
return func(value)
apply(greet, "Bob") # "Hello, Bob"
例子3:函数可以返回另一个函数
def make_multiplier(factor):
def multiplier(x):
return x * factor
return multiplier
double = make_multiplier(2)
double(5) # 10
装饰器就是这三个能力的组合:接收一个函数,在它外面包一层新功能,返回包装后的函数。
def my_decorator(func):
def wrapper():
print("执行前")
result = func()
print("执行后")
return result
return wrapper
def say_hello():
print("Hello!")
say_hello()
# 执行前
# Hello!
# 执行后
@my_decorator 等价于 say_hello = my_decorator(say_hello)。就是把 say_hello 放进 my_decorator 里包装一下,再赋值回来。
基础装饰器:给函数加”前后置”逻辑
最常用的场景:在函数执行前后自动做点事——记日志、计时、权限检查。
import time
import functools
def timer(func):
# 后面会说为什么一定要加这行
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
print(f"{func.__name__} 耗时: {elapsed:.4f}s")
return result
return wrapper
def slow_task():
time.sleep(1)
return "完成"
slow_task() # slow_task 耗时: 1.0002s
*args, **kwargs 是为了让装饰器能接受任意参数的函数——不管你传什么,我都原样传进去。
为什么一定要加 @functools.wraps
不加的话,被装饰函数的”身份信息”会丢失:
def bad_decorator(func):
def wrapper():
return func()
return wrapper
def greet():
"""打招呼"""
pass
greet.__name__ # 'wrapper',不是 'greet'
greet.__doc__ # None,文档丢了
调试的时候,堆栈里全是 wrapper,根本不知道是哪个函数出问题。用 @functools.wraps 解决:
import functools
def good_decorator(func):
# 这行把 greet 的名字、文档、签名都复制到 wrapper 上
def wrapper():
return func()
return wrapper
def greet():
"""打招呼"""
pass
greet.__name__ # 'greet'
greet.__doc__ # '打招呼'
写装饰器的铁律:每个装饰器里都用 @functools.wraps。
带参数的装饰器:三层嵌套
上面的装饰器都是”固定行为”——计时就是计时,不让你配置。
如果需要配置参数,比如”重试3次”、”重试间隔2秒”,就需要三层嵌套:
def retry(max_attempts=3, delay=1):
def decorator(func):
def wrapper(*args, **kwargs):
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_attempts:
raise
print(f"第{attempt}次失败,{delay}秒后重试...")
time.sleep(delay)
return wrapper
return decorator
def unstable_request():
# 可能失败的操作
pass
三层各管一摊:
| 层级 | 作用 | 参数 |
|---|---|---|
最外层 retry |
接收配置参数 | max_attempts、delay |
中间层 decorator |
接收被装饰的函数 | func |
最内层 wrapper |
执行实际逻辑 | *args, **kwargs |
这个模式看着复杂,但写多了就熟了——外层接配置,中层接函数,内层接调用参数。
实际场景:这几种装饰器最常用
场景1:计时器(监控接口性能)
def query_database():
# 慢查询
pass
API 响应慢的时候,加上 @timer 就能知道每个函数耗时多少,快速定位瓶颈。
场景2:重试(调用外部服务)
def call_third_party_api():
# 网络请求可能失败
pass
外部服务不稳定的时候,自动重试比人工介入快多了。
场景3:缓存(避免重复计算)
def expensive_compute(n):
# 耗时计算
pass
functools.lru_cache 是 Python 内置的缓存装饰器,同一个参数只算一次。
场景4:权限检查(Web 开发)
def dashboard():
# 只有登录用户能访问
pass
Flask、FastAPI 里这类装饰器随处可见。
场景5:日志(调试生产问题)
def process_order(order_id):
# 自动记录入参和返回值
pass
出问题的时候看日志就知道”哪个函数被调了、传了什么参数、返回了什么”。
多个装饰器叠加:顺序怎么算
def func():
pass
# 等价于:
func = decorator_a(decorator_b(decorator_c(func)))
执行顺序:从下往上包装,从上往下执行。
def work():
pass
# work() 执行时:
# log 的前置 → timer 的前置 → work → timer 的后置 → log 的后置
想清楚包装顺序再叠加,不然行为可能跟你预期的不一样。
类装饰器:需要状态的时候用
函数装饰器每次调用都是独立的,如果需要”累计调用次数”这种状态,用类装饰器。
class CountCalls:
def __init__(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)
def greet():
print("Hello")
greet() # 调用了 1 次
greet() # 调用了 2 次
greet() # 调用了 3 次
__call__ 让类的实例可以像函数一样被调用。CountCalls 的实例 greet 被调用时,走的是 __call__ 方法。
异步装饰器:用于 async def
如果你的函数是 async def,装饰器也必须是异步的:
def async_timer(func):
async def wrapper(*args, **kwargs):
start = time.perf_counter()
result = await func(*args, **kwargs) # 注意这里用 await
elapsed = time.perf_counter() - start
print(f"{func.__name__} 耗时: {elapsed:.4f}s")
return result
return wrapper
async def fetch():
await asyncio.sleep(1)
return "数据"
如果装饰器想同时支持同步和异步,用 inspect.iscoroutinefunction(func) 判断,写两个分支。
一个完整的实战:带超时的重试装饰器
import asyncio
import functools
import time
def retry_with_timeout(max_attempts=3, timeout=5):
"""
重试装饰器,支持超时控制
"""
def decorator(func):
async def wrapper(*args, **kwargs):
for attempt in range(1, max_attempts + 1):
try:
return await asyncio.wait_for(
func(*args, **kwargs),
timeout=timeout
)
except asyncio.TimeoutError:
print(f"第{attempt}次超时")
if attempt == max_attempts:
raise
except Exception as e:
print(f"第{attempt}次失败: {e}")
if attempt == max_attempts:
raise
await asyncio.sleep(1)
return wrapper
return decorator
async def fetch_data():
# 可能超时或失败的操作
await asyncio.sleep(3) # 模拟慢请求
return "数据"
这个装饰器解决了一个真实问题:外部 API 可能超时,也可能临时报错,自动重试+超时控制比手工处理可靠得多。