Python 装饰器高级用法与实战技巧
装饰器(Decorator)是 Python 中最优雅的特性之一,但很多开发者只停留在使用 @staticmethod 或 @property 的层面。本文将深入探讨装饰器的高级用法,帮助你写出更优雅、更高效的代码。
一、装饰器基础回顾
装饰器本质上是一个接受函数作为参数,并返回一个新函数的高阶函数。它允许在不修改原函数代码的情况下,为函数添加额外的功能。
1.1 最简单的装饰器
def my_decorator(func):\n def wrapper(*args, **kwargs):\n print(f"📌 调用函数: {func.__name__}")\n result = func(*args, **kwargs)\n print(f"✅ 函数执行完毕: {func.__name__}")\n return result\n return wrapper\n\n@my_decorator\ndef greet(name):\n return f"你好,{name}!"\n\nprint(greet("小豆包"))输出:
📌 调用函数: greet\n✅ 函数执行完毕: greet\n你好,小豆包!1.2 使用 functools.wraps 保留元数据
装饰器会覆盖原函数的元数据(如 __name__、__doc__),这在调试和文档生成时会有问题:
import functools\n\ndef proper_decorator(func):\n @functools.wraps(func) # 保留原函数的元数据\n def wrapper(*args, **kwargs):\n return func(*args, **kwargs)\n return wrapper\n\n@proper_decorator\ndef calculate(x, y):\n """执行加法运算"""\n return x y\n\nprint(calculate.__name__) # 输出: calculate\nprint(calculate.__doc__) # 输出: 执行加法运算二、带参数的装饰器
当装饰器需要接受参数时,需要三层嵌套:装饰器工厂 → 装饰器 → 包装函数。
2.1 可重复执行的装饰器
import functools\n\ndef repeat(times):\n """创建一个重复执行函数多次的装饰器"""\n def decorator(func):\n @functools.wraps(func)\n def wrapper(*args, **kwargs):\n results = []\n for _ in range(times):\n result = func(*args, **kwargs)\n results.append(result)\n return results[-1] # 返回最后一次执行的结果\n return wrapper\n return decorator\n\n@repeat(times=3)\ndef expensive_calculation():\n """模拟耗时计算"""\n print("⏳ 执行计算中...")\n return 42\n\nprint(expensive_calculation())输出:
⏳ 执行计算中...\n⏳ 执行计算中...\n⏳ 执行计算中...\n422.2 可选参数的装饰器
def smart_timer(prefix="耗时"):\n """可选参数的计时装饰器"""\n def decorator(func):\n @functools.wraps(func)\n def wrapper(*args, **kwargs):\n import time\n start = time.time()\n result = func(*args, **kwargs)\n end = time.time()\n print(f"{prefix}: {end - start:.4f} 秒")\n return result\n return wrapper\n return decorator\n\n# 使用自定义前缀\n@smart_timer(prefix="处理时间")\ndef process_data():\n import time\n time.sleep(0.1)\n return "处理完成"\n\nprocess_data()三、类装饰器
装饰器不仅限于函数,类也可以作为装饰器,这在需要维护状态时特别有用。
3.1 基础类装饰器
import functools\n\nclass CountCalls:\n """统计函数调用次数的类装饰器"""\n def __init__(self, func):\n self.func = func\n self.count = 0\n\n def __call__(self, *args, **kwargs):\n self.count = 1\n print(f"🔢 函数 {self.func.__name__} 已调用 {self.count} 次")\n return self.func(*args, **kwargs)\n\n@CountCalls\ndef fetch_user(user_id):\n return f"用户 {user_id} 的数据"\n\nfetch_user(1)\nfetch_user(2)\nfetch_user(3)输出:
🔢 函数 fetch_user 已调用 1 次\n🔢 函数 fetch_user 已调用 2 次\n🔢 函数 fetch_user 已调用 3 次3.2 带参数的类装饰器
class ValidateInput:\n """输入验证装饰器"""\n def __init__(self, min_value=None, max_value=None):\n self.min_value = min_value\n self.max_value = max_value\n\n def __call__(self, func):\n @functools.wraps(func)\n def wrapper(*args, **kwargs):\n for arg in args:\n if self.min_value is not None and arg < self.min_value:\n raise ValueError(f"参数 {arg} 小于最小值 {self.min_value}")\n if self.max_value is not None and arg > self.max_value:\n raise ValueError(f"参数 {arg} 大于最大值 {self.max_value}")\n return func(*args, **kwargs)\n return wrapper\n\n@ValidateInput(min_value=0, max_value=100)\ndef calculate_score(points):\n return f"得分: {points}"\n\nprint(calculate_score(85))\nprint(calculate_score(150)) # 这里会抛出异常四、装饰器链
多个装饰器可以叠加使用,执行顺序从下往上:
import functools\nimport time\n\ndef log_execution(func):\n @functools.wraps(func)\n def wrapper(*args, **kwargs):\n print(f"🔍 开始执行: {func.__name__}")\n return func(*args, **kwargs)\n return wrapper\n\ndef measure_time(func):\n @functools.wraps(func)\n def wrapper(*args, **kwargs):\n start = time.time()\n result = func(*args, **kwargs)\n end = time.time()\n print(f"⏱️ 执行时间: {end - start:.2f} 秒")\n return result\n return wrapper\n\ndef cache_result(func):\n """简单的缓存装饰器"""\n cache = {}\n @functools.wraps(func)\n def wrapper(*args, **kwargs):\n key = str(args) str(kwargs)\n if key not in cache:\n cache[key] = func(*args, **kwargs)\n print("💾 缓存未命中,执行计算")\n else:\n print("✅ 缓存命中")\n return cache[key]\n return wrapper\n\n@log_execution\n@measure_time\n@cache_result\ndef fibonacci(n):\n """计算斐波那契数列"""\n if n <= 1:\n return n\n return fibonacci(n-1) fibonacci(n-2)\n\nprint(f"fibonacci(10) = {fibonacci(10)}")\nprint(f"fibonacci(10) = {fibonacci(10)}") # 第二次会从缓存读取五、实战应用场景
5.1 API 请求限流
import functools\nimport time\nfrom datetime import datetime\n\nclass RateLimiter:\n """API 请求限流装饰器"""\n def __init__(self, calls_per_minute=60):\n self.calls_per_minute = calls_per_minute\n self.call_times = []\n\n def __call__(self, func):\n @functools.wraps(func)\n def wrapper(*args, **kwargs):\n current_time = time.time()\n # 移除 1 分钟前的调用记录\n self.call_times = [\n t for t in self.call_times\n if current_time - t < 60\n ]\n # 检查是否超过限制\n if len(self.call_times) >= self.calls_per_minute_minute:\n wait_time = 60 - (current_time - self.call_times[0])\n print(f"⏳ 请求限流,等待 {wait_time:.1f} 秒...")\n time.sleep(wait_time)\n # 记录调用时间\n self.call_times.append(current_time)\n return func(*args, **kwargs)\n return wrapper\n\n@RateLimiter(calls_per_minute=3)\ndef api_request(endpoint):\n """模拟 API 请求"""\n timestamp = datetime.now().strftime("%H:%M:%S")\n print(f"📡 API 请求到 {endpoint} [{timestamp}]")\n return {"status": "success"}\n\n# 测试限流\nfor i in range(5):\n api_request(f"/api/data/{i}")5.2 自动重试机制
import functools\nimport time\nimport random\n\ndef retry(max_attempts=3, delay=1, backoff=2):\n """自动重试装饰器,支持指数退避"""\n def decorator(func):\n @functools.wraps(func)\n def wrapper(*args, **kwargs):\n attempts = 0\n current_delay = delay\n while attempts < max_attempts:\n try:\n return func(*args, **kwargs)\n except Exception as e:\n attempts = 1\n if attempts >= max_attempts:\n print(f"❌ 重试 {max_attempts} 次后失败: {e}")\n raise\n print(f"⚠️ 第 {attempts} 次失败: {e},{current_delay} 秒后重试...")\n time.sleep(current_delay)\n current_delay *= backoff\n return wrapper\n return decorator\n\n@retry(max_attempts=3, delay=1, backoff=2)\ndef unstable_service():\n """模拟不稳定的服务"""\n if random.random() < 0.7:\n raise ConnectionError("网络连接失败")\n return {"data": "成功获取数据"}\n\nprint(unstable_service())5.3 异步装饰器
import functools\nimport asyncio\n\nasync def async_timer(func):\n """异步函数计时装饰器"""\n @functools.wraps(func)\n async def wrapper(*args, **kwargs):\n start = asyncio.get_event_loop().time()\n result = await func(*args, **kwargs)\n end = asyncio.get_event_loop().time()\n print(f"⏱️ 异步函数 {func.__name__} 耗时: {end - start:.2f} 秒")\n return result\n return wrapper\n\n@async_timer\nasync def async_task():\n """模拟异步任务"""\n await asyncio.sleep(1)\n return "任务完成"\n\n# 运行异步任务\nasyncio.run(async_task())六、性能优化技巧
6.1 使用缓存避免重复计算
import functools\n\ndef memoize(max_size=128):\n """带带大小限制的缓存装饰器"""\n def decorator(func):\n cache = {}\n order = [] # 维护访问顺序\n\n @functools.wraps(func)\n def wrapper(*args, **kwargs):\n key = str(args) str(sorted(kwargs.items()))\n if key in cache:\n # 缓存命中\n order.remove(key)\n order.append(key)\n return cache[key]\n else:\n # 缓存未命中\n result = func(*args, **kwargs)\n if len(cache) >= max_size:\n oldest_key = order.pop(0)\n del cache[oldest_key]\n cache[key] = result\n order.append(key)\n return result\n return wrapper\n return decorator\n\n@memoize(max_size=3)\ndef heavy_computation(n):\n """模拟重计算"""\n print(f"🔥 计算中: {n}")\n import time\n time.sleep(0.1)\n return n * n\n\nprint(heavy_computation(5))\nprint(heavy_computation(10))\nprint(heavy_computation(5)) { # 从缓存读取\nprint(heavy_computation(15))\nprint(heavy_computation(10)) # 从缓存读取6.2 延迟加载装饰器
import functools\n\ndef lazy_property(func):\n """延迟加载属性,只在第一次访问时计算"""\n attr_name = f"_lazy_{func.__name__}"\n\n @property\n @functools.wraps(func)\n def wrapper(self):\n if not hasattr(self, attr_name):\n setattr(self, attr_name, func(self))\n return getattr getattr(self, attr_name)\n return wrapper\n\nclass DataProcessor:\n def __init__(self, data):\n self.data = data\n\n @lazy_property\n def processed_data(self):\n print("🔄 开始处理数据...")\n # 模拟耗时操作\n import time\n time.sleep(0.5)\n return [x * 2 for x in self.data]\n\nprocessor = DataProcessor([1, 2, 3])\nprint("对象创建完成")\nprint(f"第一次访问: {processor.processed_data}") # 触发计算\nprint(f"第二次访问: {processor.processed_data}") # 使用缓存七、最佳实践建议
- 始终使用 functools.wraps:保留原函数的元数据,便于调试
- 装饰器应该轻量级:不要在装饰器中执行耗时操作
- 考虑线程安全:如果装饰器维护状态,需要考虑并发访问
- 提供清晰的错误信息:装饰器捕获异常时,保留原始异常信息
- 文档化你的装饰器:为装饰器添加详细的 docstring
八、总结
Python 装饰器是一个强大的工具,它能够:
- ✅ 横切关注点的分离(日志、缓存、限流)
- ✅ 代码复用和模块化
- ✅ 运行时行为修改
- ✅ 优雅地增强函数功能
掌握装饰器的高级用法,将让你的 Python 代码更加优雅、高效和可维护。从简单的计时器到复杂的缓存系统,装饰器都能帮你简化代码结构,提升开发效率。
进一步学习:
- 尝试编写一个上下文管理器装饰器
- 探索装饰器在 Web 框架中的应用(如 Flask、Django)
- 学习使用 wrapt 库处理复杂的装饰器场景
