Python 装饰器实战:从基础到高级应用的完整指南
装饰器是 Python 中最优雅也最强大的特性之一。它允许你在不修改原函数代码的前提下,动态地添加功能。本文将带你从装饰器的基础概念出发,逐步掌握其在实际开发中的高级应用技巧。
许多初学者对装饰器感到困惑,觉得它神秘难懂。实际上,装饰器的核心思想非常简单:它是一个接收函数作为参数并返回新函数的可调用对象。理解这一点后,你就能解锁 Python 元编程的大门。
一、装饰器的本质
让我们从最基础的例子开始。装饰器本质上是一个高阶函数,它接受一个函数作为输入,并返回一个新的函数。这个新模式让我们能够在不触碰原有代码的情况下,为函数添加额外的行为。
下面是一个简单的装饰器示例,用于测量函数执行时间:
import time\nfrom functools import wraps\n\ndef timing_decorator(func):\n """测量函数执行时间的装饰器"""\n @wraps(func)\n def wrapper(*args, **kwargs):\n start = time.perf_counter()\n result = func(*args, **kwargs)\n end = time.perf_counter()\n print(f"{func.__name__} 执行耗时:{end - start:.4f} 秒")\n return result\n return wrapper\n\n@timing_decorator\ndef slow_operation():\n """模拟耗时操作"""\n time.sleep(1)\n return "完成"\n\n# 测试\nresult = slow_operation()\nprint(f"结果:{result}")这个例子展示了装饰器的基本结构。@wraps 装饰器用于保留原函数的元数据(如函数名、文档字符串等),这是一个很好的实践习惯。
二、带参数的装饰器
实际开发中,我们经常需要给装饰器本身传递参数。这需要再嵌套一层函数,形成三层嵌套结构:
from functools import wraps\n\ndef retry(max_attempts=3, delay=1):\n """失败重试装饰器,可配置重试次数和延迟"""\n def decorator(func):\n @wraps(func)\n def wrapper(*args, **kwargs):\n attempts = 0\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 raise\n print(f"第 {attempts} 次重试,等待 {delay} 秒...")\n time.sleep(delay)\n return wrapper\n return decorator\n\n@retry(max_attempts=5, delay=0.5)\ndef unstable_api():\n """模拟不稳定的 API 调用"""\n import random\n if random.random() < 0.7:\n raise ConnectionError("网络错误")\n return "API 调用成功"\n\n# 测试\ntry:\n result = unstable_api()\n print(result)\nexcept Exception as e:\n print(f"最终失败:{e}")这个重试装饰器在实际项目中非常实用,特别是处理网络请求或数据库连接时。
三、类装饰器
装饰器不仅可以是函数,也可以是类。类装饰器通过实现 __call__ 方法来获得可调用性:
from functools import wraps\nimport time\n\nclass RateLimiter:\n """基于类的速率限制装饰器"""\n \n def __init__(self, calls_per_second=1):\n self.min_interval = 1.0 / calls_per_second\n self.last_call = 0\n \n def __call__(self, func):\n @wraps(func)\n def wrapper(*args, **kwargs):\n elapsed = time.time() - self.last_call\n if elapsed < self.min_interval:\n sleep_time = self.min_interval - elapsed\n time.sleep(sleep_time)\n self.last_call = time.time()\n return func(*args, **kwargs)\n return wrapper\n\n@RateLimiter(calls_per_second=2)\ndef api_call():\n """限制每秒最多调用 2 次"""\n print(f"API 调用于 {time.time():.2f}")\n return "OK"\n\n# 测试:快速调用多次\nfor i in range(5):\n api_call()类装饰器的优势在于可以维护状态,适合需要记住之前调用信息的场景。
四、装饰器栈与执行顺序
一个函数可以应用多个装饰器,形成装饰器栈。关键是要理解它们的执行顺序:从内到外应用,从外到内执行。
from functools import wraps\n\ndef decorator_a(func):\n @wraps(func)\n def wrapper(*args, **kwargs):\n print("[装饰器 A] 前置处理")\n result = func(*args, **kwargs)\n print("[装饰器 A] 后置处理")\n return result\n return wrapper\n\ndef decorator_b(func):\n @wraps(func)\n def wrapper(*args, **kwargs):\n print(" [装饰器 B] 前置处理")\n result = func(*args, **kwargs)\n print(" [装饰器 B] 后置处理")\n return result\n return wrapper\n\n@decorator_a\n@decorator_b\ndef target_function():\n print(" >>> 目标函数执行 <<<")\n return "结果"\n\nprint("=== 执行开始 ===")\ntarget_function()\nprint("=== 执行结束 ===")运行这段代码,你会看到装饰器 B 先被应用(内层),但执行时装饰器 A 的外层逻辑先运行。理解这个顺序对调试复杂装饰器链非常重要。
五、实战:构建日志装饰器
让我们综合运用所学知识,创建一个生产级别的日志装饰器:
from functools import wraps\nimport logging\nimport time\nfrom datetime import datetime\n\n# 配置日志\nlogging.basicConfig(\n level=logging.INFO,\n format='%(asctime)s - %(levelname)s - %(message)s'\n)\nlogger = logging.getLogger(__name__)\n\ndef log_execution(log_level=logging.INFO, include_args=True, include_result=True):\n """\n 生产级日志装饰器\n \n 参数:\n log_level: 日志级别\n include_args: 是否记录参数\n include_result: 是否记录返回值\n """\n def decorator(func):\n @wraps(func)\n def wrapper(*args, **kwargs):\n func_name = func.__name__\n start_time = time.time()\n \n # 构建日志消息\n log_parts = [f"调用函数:{func_name}"]\n \n if include_args:\n arg_str = ", ".join(\n [repr(a) for a in args] + \n [f"{k}={repr(v)}" for k, v in kwargs.items()]\n )\n log_parts.append(f"参数:[{arg_str}]")\n \n logger.log(log_level, " | ".join(log_parts))\n \n try:\n result = func(*args, **kwargs)\n elapsed = time.time() - start_time\n \n result_parts = [f"函数 {func_name} 执行完成"]\n if include_result:\n result_parts.append(f"返回值:{repr(result)}")\n result_parts.append(f"耗时:{elapsed:.4f}秒")\n \n logger.log(log_level, " | ".join(result_parts))\n return result\n \n except Exception as e:\n elapsed = time.time() - start_time\n logger.error(\n f"函数 {func_name} 执行失败 | 错误:{type(e).__name__}: {e} | 耗时:{elapsed:.4f}秒"\n )\n raise\n \n return wrapper\n return decorator\n\n# 使用示例\n@log_execution(include_args=True, include_result=False)\ndef calculate_sum(a, b, c=0):\n """计算多个数的和"""\n return a + b + c\n\n@log_execution(log_level=logging.DEBUG)\ndef divide(x, y):\n """除法运算"""\n return x / y\n\n# 测试\nprint("\n=== 测试 1:正常调用 ===")\ncalculate_sum(10, 20, c=5)\n\nprint("\n=== 测试 2:异常处理 ===")\ntry:\n divide(10, 0)\nexcept ZeroDivisionError:\n print("捕获到除零错误")这个日志装饰器展示了多个高级特性:可配置的日志级别、可选的参数/结果记录、异常处理和执行时间统计。在实际项目中,这样的装饰器可以大大简化调试和监控工作。
六、注意事项与最佳实践
使用装饰器时,有几点需要特别注意:
1. 始终使用 @wraps:这能保留原函数的 __name__、__doc__ 等属性,对调试和文档生成至关重要。
2. 避免过度嵌套:如果装饰器逻辑太复杂,考虑拆分成多个小装饰器或使用类来实现。
3. 注意性能影响:每个装饰器都会增加函数调用的开销,在性能敏感的场景要谨慎使用。
4. 文档化装饰器行为:在装饰器的 docstring 中清楚说明它的作用、参数和使用场景。
总结
装饰器是 Python 元编程的基石之一。掌握它不仅能让你写出更优雅的代码,还能帮助你理解许多流行框架(如 Flask、FastAPI)的内部机制。从今天开始,尝试在你的项目中应用这些技巧,你会发现代码变得更加模块化和可维护。
记住:好的装饰器应该是透明的、可组合的,并且有明确的单一职责。遵循这些原则,你就能充分发挥装饰器的威力。
