Python 装饰器深度解析:从原理到实战
装饰器是 Python 中最优雅的语言特性之一,它让我们能够在不修改原函数代码的情况下,动态地增强函数的功能。本文将深入讲解装饰器的底层原理、多种高级用法,并通过实际案例展示如何在项目中应用装饰器。
一、装饰器的本质:闭包与高阶函数
从本质上说,装饰器就是一个接收函数作为参数,并返回一个新函数的高阶函数。它的实现依赖于闭包这一机制。
def timer_decorator(func):
"""记录函数执行时间的装饰器"""
import time
import functools
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
end = time.perf_counter()
print(f"{func.__name__} 执行耗时: {end - start:.4f} 秒")
return result
return wrapper
@timer_decorator
def calculate_sum(n):
"""计算 1 到 n 的和"""
total = sum(range(n + 1))
return total
# 调用函数
result = calculate_sum(1000000)
print(f"计算结果: {result}")
这里使用了 functools.wraps 来保留原函数的元信息(如 __name__、__doc__ 等),这是一个重要的最佳实践。
二、带参数的装饰器工厂
当我们需要为装饰器传递配置参数时,可以使用装饰器工厂模式。这实际上是一个返回装饰器的函数。
def retry(max_attempts=3, delay=1):
"""失败重试装饰器工厂"""
import time
import functools
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
import random
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_attempts - 1:
raise
print(f"第 {attempt + 1} 次尝试失败: {e},{delay} 秒后重试...")
time.sleep(delay)
return None
return wrapper
return decorator
@retry(max_attempts=3, delay=0.5)
def unreliable_api_call():
"""模拟可能失败的 API 调用"""
import random
if random.random() < 0.7: # 70% 概率失败
raise ConnectionError("网络连接失败")
return {"status": "success", "data": "API 返回数据"}
# 调用函数
try:
result = unreliable_api_call()
print(f"API 调用成功: {result}")
except Exception as e:
print(f"API 调用最终失败: {e}")
三、类装饰器
除了函数装饰器,Python 还支持类装饰器。类装饰器通过实现 __call__ 方法,使类的实例可以像函数一样被调用。
class CacheDecorator:
"""缓存装饰器类"""
def __init__(self, func):
self.func = func
self.cache = {}
def __call__(self, *args, **kwargs):
# 使用参数作为缓存键
cache_key = (args, frozenset(kwargs.items()))
if cache_key not in self.cache:
print(f"计算 {self.func.__name__}({args[0]})...")
self.cache[cache_key] = self.func(*args, **kwargs)
else:
print(f"从缓存获取 {self.func.__name__}({args[0]})...")
return self.cache[cache_key]
@CacheDecorator
def fibonacci(n):
"""计算斐波那契数列"""
if n <= 1:
return n
return fibonacci(n - 1) + fibonacci(n - 2)
# 调用函数
print(f"fibonacci(10) = {fibonacci(10)}") # 会计算
print(f"fibonacci(10) = {fibonacci(10)}") # 从缓存获取
print(f"fibonacci(15) = {fibonacci(15)}") # 会计算(部分结果已缓存)
四、多个装饰器的链式调用
Python 允许为一个函数应用多个装饰器,装饰器的执行顺序是从下到上(靠近函数的装饰器先执行)。
def log_before(func):
"""执行前记录日志"""
def wrapper(*args, **kwargs):
print(f">>> 即将执行 {func.__name__},参数: {args}, {kwargs}")
return func(*args, **kwargs)
return wrapper
def log_after(func):
"""执行后记录日志"""
def wrapper(*args, **kwargs):
result = func(*args, **kwargs)
print(f"<<< {func.__name__} 执行完成,返回值: {result}")
return result
return wrapper
@log_after
@log_before
def multiply(a, b):
"""两个数相乘"""
return a * b
# 调用函数
result = multiply(5, 3)
print(f"最终结果: {result}")
五、实战案例:权限验证装饰器
在实际开发中,装饰器常用于权限验证、日志记录、性能监控等场景。下面实现一个实用的权限验证装饰器。
class PermissionDeniedError(Exception):
"""权限不足异常"""
pass
def require_role(*allowed_roles):
"""权限验证装饰器"""
def decorator(func):
def wrapper(user, *args, **kwargs):
if user.get('role') not in allowed_roles:
raise PermissionDeniedError(
f"用户 {user.get('name')} 无权限执行 {func.__name__},"
f"需要角色: {allowed_roles},当前角色: {user.get('role')}"
)
return func(user, *args, **kwargs)
return wrapper
return decorator
# 模拟用户数据
admin_user = {'name': '张三', 'role': 'admin'}
normal_user = {'name': '李四', 'role': 'user'}
guest_user = {'name': '王五', 'role': 'guest'}
@require_role('admin')
def delete_user(operator, target_user_id):
"""删除用户(仅管理员)"""
return f"用户 {target_user_id} 已被 {operator['name']} 删除"
@require_role('admin', 'user')
def update_profile(user, user_id, new_data):
"""更新个人资料(管理员和普通用户)"""
return f"用户 {user_id} 的资料已更新"
# 测试权限验证
try:
print(delete_user(admin_user, 1001))
except PermissionDeniedError as e:
print(f"错误: {e}")
try:
print(delete_user(normal_user, 1001))
except PermissionDeniedError as e:
print(f"错误: {e}")
try:
print(update_profile(normal_user, 1002, {'age': 25}))
except PermissionDeniedError as e:
print(f"错误: {e}")
六、装饰器性能监控应用
import time
import functools
from collections import defaultdict
class PerformanceMonitor:
"""性能监控器"""
def __init__(self):
self.call_counts = defaultdict(int)
self.total_times = defaultdict(float)
def track(self, func):
"""性能跟踪装饰器"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
end = time.perf_counter()
self.call_counts[func.__name__] += 1
self.total_times[func.__name__] += (end - start)
return result
return wrapper
def report(self):
"""生成性能报告"""
print("\n=== 性能统计报告 ===")
for func_name in sorted(self.call_counts.keys()):
count = self.call_counts[func_name]
total_time = self.total_times[func_name]
avg_time = total_time / count if count > 0 else 0
print(
f"{func_name:20s} | 调用次数: {count:4d} | "
f"总耗时: {total_time:8.4f}s | 平均: {avg_time:8.4f}s"
)
# 创建监控器实例
monitor = PerformanceMonitor()
@monitor.track
def heavy_computation(n):
"""模拟耗时计算"""
total = 0
for i in range(n):
total += i ** 2
return total
@monitor.track
def quick_lookup(data, key):
"""模拟快速查询"""
return data.get(key, None)
# 执行一些操作
for _ in range(5):
heavy_computation(10000)
for _ in range(100):
quick_lookup({'a': 1, 'b': 2}, 'a')
# 生成性能报告
monitor.report()
七、总结
Python 装饰器是一个强大的工具,它让我们能够:
1. 实现关注点分离:将日志、权限验证、性能监控等横切关注点从业务逻辑中分离出来
2. 提高代码复用性:一个装饰器可以应用到多个函数上
3. 保持代码整洁:避免在每个函数中重复相同的代码
4. 动态增强功能:在不修改原函数的情况下添加新功能
掌握装饰器不仅能让代码更加优雅,还能显著提升开发效率。建议在实际项目中多思考哪些场景适合使用装饰器,逐步将其融入到你的代码风格中。
记住:装饰器的核心思想是"函数也是对象,可以像变量一样传递和返回"。理解这一点,你就掌握了装饰器的精髓。
