Python装饰器完全指南:从原理到实战
Python 装饰器(Decorator)是 Python 中最强大也是最优雅的特性之一。它允许你在不修改原函数代码的情况下,动态地给函数添加功能。这种设计模式体现了 AOP(面向切面编程)的思想,让代码更加清晰、可维护。
装饰器的基本原理
装饰器本质上是一个接受函数作为参数,并返回一个新函数的高阶函数。当使用 @decorator 语法时,Python 会自动将下方的函数传递给装饰器函数,并将返回的函数替换原函数。这个过程发生在函数定义时,而非调用时。
让我们从一个最简单的计时装饰器开始:
import time import functools from typing import Callable, Any def timer(func: Callable) -> Callable: """记录函数执行时间的装饰器""" @functools.wraps(func) def wrapper(*args, **kwargs) -> Any: start_time = time.time() result = func(*args, **kwargs) end_time = time.time() print(f"{func.__name__} 执行耗时: {end_time - start_time:.4f} 秒") return result return wrapper # 使用装饰器 @timer def calculate_sum(n: int) -> int: """计算 1 到 n 的和""" total = 0 for i in range(1, n 1): total = i time.sleep(0.01) # 模拟耗时操作 return total result = calculate_sum(100) print(f"计算结果: {result}") 运行这段代码,你会看到类似这样的输出:
calculate_sum 执行耗时: 1.0105 秒 计算结果: 5050 为什么需要 functools.wraps?
使用 @functools.wraps 装饰器内部函数非常重要。它会将原函数的 __name__、__doc__、__module__ 等属性复制到 wrapper 函数中,避免调试时的困惑。如果不使用它,打印 calculate_sum.__name__ 会得到 "wrapper" 而不是 "calculate_sum"。
带参数的装饰器
有时候我们需要装饰器接受自定义参数,比如指定重试次数、缓存大小等。这时需要创建一个返回装饰器的工厂函数:
def repeat(times: int = 3, delay: float = 0.1) -> Callable: """多次执行函数,返回最后一次结果""" def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(*args, **kwargs) -> Any: results = [] for i in range(times): print(f"第 {i 1}/{times} 次执行 {func.__name__}...") result = func(*args, **kwargs) results.append(result) if i < times - 1 and delay > 0: time.sleep(delay) return results[-1] if results else None return wrapper return decorator @repeat(times=2, delay=0.05) def fetch_data(url: str) -> str: """模拟从 API 获取数据""" print(f"从 {url} 获取数据...") time.sleep(0.2) return "模拟数据" data = fetch_data("https://api.example.com/data") print(f"获取的数据: {data}") 实战应用:缓存装饰器
缓存是装饰器最常见的应用场景之一。对于计算密集型或需要频繁调用的函数,缓存可以大幅提升性能:
def memoize(max_size: int = 128) -> Callable: """带大小限制的缓存装饰器""" def decorator(func: Callable) -> Callable: cache = {} cache_order = [] @functools.wraps(func) def wrapper(*args, **kwargs) -> Any: # 创建缓存键 key = str(args) str(sorted(kwargs.items())) # 缓存命中 if key in cache: print(f"缓存命中: {key[:50]}...") return cache[key] # 计算并缓存结果 result = func(*args, **kwargs) cache[key] = result cache_order.append(key) # 限制缓存大小(LRU) if len(cache) > max_size: oldest_key = cache_order.pop(0) del cache[oldest_key] return result return wrapper return decorator @memoize(max_size=50) def fibonacci(n: int) -> int: """计算斐波那契数列(使用缓存优化)""" if n <= 1: return n return fibonacci(n-1) fibonacci(n-2) # 测试缓存效果 print("计算 fibonacci(30)...") start = time.time() result = fibonacci(30) print(f"结果: {result}, 耗时: {time.time() - start:.4f}秒") print("\n再次计算 fibonacci(30)...") start = time.time() result = fibonacci(30) print(f"结果: {result}, 耗时: {time.time() - start:.4f}秒") 异常处理装饰器
统一的异常处理是另一个重要应用场景。它可以避免在多个函数中重复 try-except 代码:
def handle_errors( default_return: Any = None, exceptions: tuple = (Exception,), log_error: bool = True ) -> Callable: """异常处理装饰器""" def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(*args, **kwargs) -> Any: try: return func(*args, **kwargs) except exceptions as e: if log_error: print(f"[错误] {func.__name__}: {type(e).__name__}: {e}") return default_return return wrapper return decorator @handle_errors(default_return=None, exceptions=(ZeroDivisionError, ValueError)) def safe_divide(a: int, b: int) -> float: """安全除法""" return a / b @handle_errors(default_return="操作失败", log_error=True) def parse_json(text: str) -> dict: """解析 JSON 字符串""" return json.loads(text) print(safe_divide(10, 2)) # 5.0 print(safe_divide(10, 0)) # None,不报错 print(parse_json('{"name": "张三"}')) # {'name': '张三'} print(parse_json('invalid json')) # 操作失败 装饰器链
多个装饰器可以叠加使用,执行顺序从内到外(从下到上):
@timer @memoize(max_size=32) def expensive_calculation(x: int) -> int: """耗时的计算函数""" print(f"计算 {x} 的三次方...") time.sleep(0.1) return x ** 3 # 等价于: # expensive_calculation = timer(memoize(max_size=32)(expensive_calculation)) print(expensive_calculation(5)) print(expensive_calculation(5)) # 第二次会命中缓存 类装饰器
装饰器不仅可以用于函数,也可以用于类。类装饰器通常用于动态添加方法或属性:
def add_repr(cls) -> type: """为类添加 __repr__ 方法""" original_init = cls.__init__ def __init__(self, *args, **kwargs): original_init(self, *args, **kwargs) self._init_args = args self._init_kwargs = kwargs def __repr__(self) -> str: args_str = ", ".join(repr(arg) for arg in self._init_args) kwargs_str = ", ".join(f"{k}={v!r}" for k, v in self._init_kwargs.items()) all_args = ", ".join(filter(None, [args_str, kwargs_str])) return f"{self.__class__.__name__}({all_args})" cls.__init__ = __init__ cls.__repr__ = __repr__ return cls @add_repr class Person: def __init__(self, name: str, age: int, city: str = "北京"): self.name = name self.age = age self.city = city person = Person("李四", 30, "上海") print(person) # Person('李四', 30, city='上海') 性能监控装饰器
在生产环境中,性能监控非常重要。下面是一个实用的性能监控装饰器:
class PerformanceMonitor: """性能监控器""" def __init__(self): self.stats = {} def record(self, func: Callable) -> Callable: """记录函数调用统计信息""" @functools.wraps(func) def wrapper(*args, **kwargs) -> Any: func_name = func.__name__ if func_name not in self.stats: self.stats[func_name] = { 'calls': 0, 'total_time': 0, 'errors': 0 } self.stats[func_name]['calls'] = 1 start_time = time.time() try: result = func(*args, **kwargs) elapsed = time.time() - start_time self.stats[func_name]['total_time'] = elapsed return result except Exception: self.stats[func_name]['errors'] = 1 raise return wrapper def report(self) -> None: """打印性能报告""" print("\n=== 性能监控报告 ===") for func_name, stats in self.stats.items(): avg_time = stats['total_time'] / stats['calls'] if stats['calls'] > 0 else 0 print(f"{func_name}:") print(f" 调用次数: {stats['calls']}") print(f" 总耗时: {stats['total_time']:.4f}s") print(f" 平均耗时: {avg_time:.4f}s") print(f" 错误次数: {stats['errors']}") # 使用示例 monitor = PerformanceMonitor() @monitor.record def process_data(items: list) -> list: """处理数据""" time.sleep(0.05) return [item * 2 for item in items] @monitor.record def validate_data(data: dict) -> bool: """验证数据""" time.sleep(0.02) return 'id' in data # 测试 for i in range(5): process_data([1, 2, 3, 4, 5]) validate_data({'id': i}) monitor.report() 最佳实践
1. 始终使用 @functools.wraps 保留原函数的元数据
2. 装饰器应该保持透明,不改变原函数的核心行为
3. 避免过度使用装饰器,保持代码可读性
4. 装饰器内部函数使用类型注解,提高代码质量
5. 对于复杂的装饰器,考虑使用类实现而不是嵌套函数
6. 文档化装饰器的参数和行为
总结
Python 装饰器是一种强大的工具,可以优雅地实现横切关注点,如日志记录、性能监控、缓存、异常处理等。掌握装饰器将让你的 Python 代码更加简洁、高效和可维护。从简单的单层装饰器开始,逐步掌握带参数的装饰器、装饰器链和类装饰器,你将在实际项目中充分发挥装饰器的威力。
记住:装饰器的核心价值在于在不修改原代码的情况下动态增强功能,这正是 Python 之"优雅"的体现。
