Python 装饰器高级实战:从基础到精通的5个实用技巧
引言:为什么要深入掌握装饰器?
装饰器是 Python 中最优雅的元编程工具之一,它能在不修改原函数代码的情况下,动态地增加功能。很多开发者都知道如何使用 @timer 计时或 @cache 缓存,但装饰器的真正威力远不止于此。
当你需要在多个函数中添加日志、权限检查、重试逻辑、性能监控等横切关注点时,装饰器能让你避免代码重复,保持业务逻辑的纯净。本文将从实战角度出发,分享 5 个高级装饰器技巧,每个都配有可直接运行的代码示例。
技巧一:带参数的装饰器工厂
基础装饰器只能接受被装饰的函数,但实际项目中,我们经常需要给装饰器传递参数。比如创建一个可配置的重试装饰器,可以指定最大重试次数和延迟时间。
import time
from functools import wraps
def retry(max_attempts=3, delay=1, exceptions=(Exception,)):
"""带参数的装饰器工厂函数"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt < max_attempts:
print(f"尝试 {attempt}/{max_attempts} 失败: {e}, {delay}秒后重试...")
time.sleep(delay)
raise Exception(f"所有 {max_attempts} 次尝试都失败了") from last_exception
return wrapper
return decorator
# 使用示例
@retry(max_attempts=3, delay=2, exceptions=(ConnectionError, TimeoutError))
def fetch_data(url):
import random
if random.random() < 0.7:
raise ConnectionError("网络连接失败")
return f"从 {url} 获取的数据"
# 测试
try:
result = fetch_data("https://api.example.com")
print(f"成功: {result}")
except Exception as e:
print(f"最终失败: {e}")
这里的 key 理解:retry() 返回真正的装饰器,decorator(func) 返回包装函数。这种三层嵌套结构是带参数装饰器的标准模式。
技巧二:类装饰器 - 更强大的状态管理
函数装饰器适合简单场景,但当装饰器需要维护复杂状态或提供额外方法时,类装饰器是更好的选择。类装饰器可以实现类似中间件的链式调用。
class AccessLogger:
"""带状态管理的类装饰器"""
def __init__(self, log_file="access.log"):
self.log_file = log_file
self.call_count = {} # 记录每个函数的调用次数
def __call__(self, func):
def wrapper(*args, **kwargs):
# 记录调用
func_name = func.__name__
self.call_count[func_name] = self.call_count.get(func_name, 0) + 1
# 记录日志
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
log_entry = f"[{timestamp}] {func_name} 被调用 (第{self.call_count[func_name]}次)\n"
with open(self.log_file, "a") as f:
f.write(log_entry)
print(f"📊 {func_name} 累计调用: {self.call_count[func_name]} 次")
return func(*args, **kwargs)
return wrapper
def get_stats(self):
"""获取统计信息"""
return self.call_count.copy()
# 使用示例
logger = AccessLogger()
@logger
def process_user_data(user_id):
return f"处理用户 {user_id} 的数据"
@logger
def send_notification(message):
return f"发送通知: {message}"
# 测试
process_user_data(1001)
process_user_data(1002)
send_notification("欢迎注册")
send_notification("密码重置")
print(f"\n统计: {logger.get_stats()}")
类装饰器的优势在于可以存储状态(如调用计数、日志信息),并且可以提供额外方法(如 get_stats())来查询这些状态。
技巧三:装饰器堆叠顺序 - 谁先执行?
当多个装饰器堆叠使用时,理解执行顺序至关重要。装饰器的执行顺序是"从下到上注册,从上到下执行"。
def auth_check(required_role="user"):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(f"🔐 权限检查: 需要 {required_role}权限")
return func(*args, **kwargs)
return wrapper
return decorator
def rate_limit(max_calls=10):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(f"⚡ 限流检查: 最多 {max_calls} 次/分钟")
return func(*args, **kwargs)
return wrapper
return decorator
def log_performance(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
print(f"📏 性能监控: 开始计时")
result = func(*args, **kwargs)
elapsed = time.time() - start
print(f"📏 性能监控: 耗时 {elapsed:.4f} 秒")
return result
return wrapper
# 装饰器堆叠 - 注意顺序!
@rate_limit(max_calls=10)
@auth_check(required_role="admin")
@log_performance
def sensitive_operation(data):
print(f"执行敏感操作: {data}")
return "操作完成"
# 执行
result = sensitive_operation("删除重要文件")
print("\n--- 执行顺序解析 ---")
print("1. rate_limit (最先执行,最外层)")
print("2. auth_check (第二层)")
print("3. log_performance (第三层)")
print("4. sensitive_operation (原函数)")
print("\n实际调用链: rate_limit → auth_check → log_performance → sensitive_operation")
实战建议:将最通用的装饰器(如日志、性能监控)放在最下面,将最具体的(如权限检查)放在最上面。这样可以让后续装饰器基于前面的装饰器结果进行决策。
技巧四:异步装饰器 - 适配 async/await
现代 Python 项目中大量使用异步编程。装饰器也需要正确支持异步函数,否则会导致 RuntimeWarning 或协程泄漏。
import asyncio
from functools import wraps
def async_cache(ttl=60):
"""支持异步函数的缓存装饰器"""
cache = {}
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# 生成缓存键
cache_key = (func.__name__, args, frozenset(kwargs.items()))
# 检查缓存
if cache_key in cache:
cached_time, result = cache[cache_key]
if time.time() - cached_time < ttl:
print(f"🎯 缓存命中: {func.__name__}")
return result
# 执行原函数
print(f"🔄 执行: {func.__name__}")
result = await func(*args, **kwargs)
# 存入缓存
cache[cache_key] = (time.time(), result)
return result
return wrapper
return decorator
def async_timeout(seconds=10):
"""异步函数超时控制装饰器"""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
try:
return await asyncio.wait_for(func(*args, **kwargs), timeout=seconds)
except asyncio.TimeoutError:
raise TimeoutError(f"{func.__name__} 执行超时 ({seconds}秒)")
return wrapper
return decorator
# 使用示例
@async_cache(ttl=5)
@async_timeout(seconds=3)
async def fetch_user_data(user_id):
print(f"查询用户 {user_id} 数据...")
await asyncio.sleep(1) # 模拟网络请求
return {"id": user_id, "name": f"User{user_id}", "email": f"user{user_id}@example.com"}
async def main():
# 第一次调用 - 会执行
print("--- 第一次调用 ---")
data1 = await fetch_user_data(1001)
print(f"结果: {data1}")
# 第二次调用 - 命中缓存
print("\n--- 第二次调用 (缓存) ---")
data2 = await fetch_user_data(1001)
print(f"结果: {data2}")
# 运行测试
asyncio.run(main())
关键点:async def wrapper 和 await func(*args, **kwargs)。注意 asyncio.wait_for() 可以优雅地处理超时,避免长时间挂起。
技巧五:装饰器性能优化 - 避免常见陷阱
装饰器虽然优雅,但使用不当会影响性能。以下是几个优化技巧:
1. 始终使用 functools.wraps
from functools import wraps
# ❌ 错误做法
def bad_decorator(func):
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
# ✅ 正确做法
def good_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
# 测试差异
@bad_decorator
def example_func():
"""这是一个示例函数"""
pass
print(f"❌ 错误做法: {example_func.__name__}, 文档: {example_func.__doc__}")
# 输出: wrapper, None
@good_decorator
def example_func2():
"""这是一个示例函数"""
pass
print(f"✅ 正确做法: {example_func2.__name__}, 文档: {example_func2.__doc__}")
# 输出: example_func2, 这是一个示例函数
2. 避免在装饰器中创建昂贵对象
# ❌ 每次调用都创建数据库连接
def bad_db_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
db = create_expensive_db_connection() # 严重性能问题!
result = func(*args, **kwargs, db=db)
db.close()
return result
return wrapper
# ✅ 使用闭包或类装饰器缓存连接
class DBConnectionPool:
_instance = None
@classmethod
def get_connection(cls):
if cls._instance is None:
cls._instance = create_expensive_db_connection()
return cls._instance
def good_db_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
db = DBConnectionPool.get_connection() # 复用连接
return func(*args, **kwargs, db=db)
return wrapper
3. 使用 __qualname__ 而非 __name__ 处理类方法
class MyClass:
@good_decorator
def method(self):
pass
# __name__ 在类方法中可能不准确
print(f"__name__: {MyClass.method.__name__}")
print(f"__qualname__: {MyClass.method.__qualname__}")
总结:装饰器的最佳实践清单
✅ 使用 functools.wraps:保留原函数的元信息(名称、文档字符串等)
✅ 支持 *args, **kwargs:让装饰器更通用
✅ 考虑异步支持:如果项目使用 asyncio,确保装饰器兼容
✅ 合理控制堆叠顺序:通用装饰器在下,具体装饰器在上
✅ 避免重复计算:使用闭包或类属性缓存昂贵对象
✅ 提供清晰文档:装饰器应该有良好的 docstring 和示例
装饰器是提升 Python 代码质量和可维护性的利器。掌握这些高级技巧后,你可以在不修改业务代码的情况下,优雅地添加日志、缓存、权限检查、性能监控等功能。记住:装饰器应该让代码更清晰,而不是更复杂。
延伸阅读:如果想进一步探索,可以研究 Python 的 contextlib 模块(上下文管理器)和 __getattr__、__getattribute__ 等魔术方法,它们与装饰器一样,都是 Python 元编程的强大工具。
