核心概念
📌 什么是装饰器?
装饰器(Decorator)是 Python 中一种强大的语法特性,它允许你在不修改原函数代码的情况下,动态地扩展函数的功能。装饰器本质上是一个接收函数作为参数并返回新函数的高阶函数。
装饰器使用 @ 符号作为语法糖,使代码更加简洁优雅。它是 Python 函数式编程和元编程的重要组成部分。
代码复用
将通用功能抽象为装饰器,避免代码重复,提高可维护性
关注点分离
将横切关注点(如日志、权限)与业务逻辑解耦
动态增强
在运行时动态修改函数行为,无需改动源代码
可组合性
多个装饰器可以叠加使用,灵活组合功能
工作原理
🔍 装饰器的本质
要理解装饰器,首先需要理解 Python 中函数是一等公民这一概念。函数可以:
- 作为参数传递给其他函数
- 作为其他函数的返回值
- 赋值给变量
- 存储在数据结构中
💡 核心理解
当你写 @decorator 放在函数定义上方时,Python
会自动将该函数作为参数传递给装饰器,并用装饰器的返回值替换原函数。这就是为什么装饰器必须返回一个可调用对象。
代码示例
🔰 示例1:基础装饰器 - 函数计时器
最基础的装饰器用法,用于测量函数执行时间:
import time
from functools import wraps
# 定义一个计时装饰器
def timer(func):
"""测量函数执行时间的装饰器"""
@wraps(func) # 保留原函数的元信息
def wrapper(*args, **kwargs):
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 slow_function():
"""模拟一个耗时操作"""
time.sleep(1)
return "完成!"
# 调用函数
result = slow_function()
print(f"返回值: {result}")
⚡ 示例2:带参数的装饰器 - 重试机制
当装饰器本身需要接收参数时,需要再嵌套一层函数:
import time
from functools import wraps
# 带参数的重试装饰器
def retry(max_attempts=3, delay=1):
"""
自动重试装饰器
:param max_attempts: 最大重试次数
:param delay: 重试间隔(秒)
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
attempts = 0
while attempts < max_attempts:
try:
return func(*args, **kwargs)
except Exception as e:
attempts += 1
print(f"第 {attempts} 次尝试失败: {e}")
if attempts < max_attempts:
print(f"等待 {delay} 秒后重试...")
time.sleep(delay)
raise Exception(f"函数 {func.__name__} 在 {max_attempts} 次尝试后仍然失败")
return wrapper
return decorator
# 模拟不稳定的网络请求
attempt_count = 0
@retry(max_attempts=3, delay=0.5)
def unstable_network_call():
"""模拟不稳定的网络请求"""
global attempt_count
attempt_count += 1
if attempt_count < 3:
raise ConnectionError("网络连接失败")
return "请求成功!"
# 测试重试机制
result = unstable_network_call()
print(f"最终结果: {result}")
🚀 示例3:类装饰器 - 缓存机制
使用类实现装饰器,可以维护状态,实现更复杂的功能:
from functools import wraps
class Memoize:
"""
缓存装饰器类
缓存函数的计算结果,避免重复计算
"""
def __init__(self, func):
self.func = func
self.cache = {} # 存储缓存结果
wraps(func)(self) # 保留原函数元信息
def __call__(self, *args):
# 检查缓存中是否已有结果
if args in self.cache:
print(f" → 缓存命中: {args}")
return self.cache[args]
# 计算结果并存入缓存
print(f" → 计算中: {args}")
result = self.func(*args)
self.cache[args] = result
return result
def clear_cache(self):
"""清空缓存"""
self.cache.clear()
# 使用缓存装饰器优化递归计算
@Memoize
def fibonacci(n):
"""计算斐波那契数列第 n 项"""
if n < 2:
return n
return fibonacci(n - 1) + fibonacci(n - 2)
# 测试缓存效果
print("计算 fibonacci(5):")
result = fibonacci(5)
print(f"结果: {result}\n")
print("再次计算 fibonacci(5):")
result = fibonacci(5)
print(f"结果: {result}")
高级应用
🎯 常见应用场景
日志记录
自动记录函数调用参数、返回值和执行时间
权限验证
在执行函数前检查用户权限,实现访问控制
结果缓存
缓存函数计算结果,避免重复计算提升性能
参数验证
在函数执行前自动验证参数类型和范围
事务管理
数据库事务的自动提交和回滚处理
限流控制
限制函数的调用频率,防止资源过度使用
最佳实践
✅ 使用要点
使用 @wraps
始终使用 functools.wraps 保留原函数的元信息(__name__, __doc__ 等)
支持所有参数
使用 *args 和 **kwargs 确保装饰器能处理任意参数的函数
保持简单
每个装饰器只做一件事,遵循单一职责原则
添加文档
为装饰器编写清晰的文档字符串说明其功能
⚠️ 常见陷阱
- 忘记调用原函数或返回其结果
- 带参数装饰器忘记额外的嵌套层
- 装饰器副作用影响了函数的纯净性
- 过度使用装饰器导致调试困难
💡 Python 标准库中的装饰器
• @staticmethod - 定义静态方法
• @classmethod - 定义类方法
• @property - 将方法转为属性访问
• @functools.lru_cache - 内置的缓存装饰器
• @dataclasses.dataclass - 自动生成类方法
性能与调试
📊 装饰器性能影响
装饰器会增加额外的函数调用层次,需要注意性能开销:
import time
from functools import wraps
# 性能测试装饰器
def benchmark(iterations=10000):
"""测试函数执行性能"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
for _ in range(iterations):
result = func(*args, **kwargs)
end = time.perf_counter()
avg_time = (end - start) / iterations * 1000 # 毫秒
print(f"{func.__name__}: {avg_time:.6f}ms 平均耗时 ({iterations}次调用)")
return result
return wrapper
return decorator
# 比较使用装饰器和不使用装饰器的性能
def simple_add(a, b):
return a + b
def heavy_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 模拟复杂的装饰器逻辑
_ = [i for i in range(100)] # 额外开销
return func(*args, **kwargs)
return wrapper
@heavy_decorator
def decorated_add(a, b):
return a + b
# 性能比较
print("性能比较:")
benchmark()(lambda: simple_add(1, 2))()
benchmark()(lambda: decorated_add(1, 2))()
⚡ 性能优化建议
1. 避免过度嵌套:多层装饰器会增加调用栈深度
2. 使用内置装饰器:如 @lru_cache 经过高度优化
3. 懒惰计算:只在必要时执行装饰器逻辑
4. 缓存结果:对于相同输入,缓存计算结果
🐞 装饰器调试技巧
装饰器可能使调试变得复杂,以下是一些实用的调试方法:
from functools import wraps
import inspect
# 调试装饰器:记录函数调用信息
def debug(func):
"""打印函数调用详细信息的装饰器"""
@wraps(func)
def wrapper(*args, **kwargs):
# 获取调用位置
frame = inspect.currentframe().f_back
filename = frame.f_code.co_filename
lineno = frame.f_lineno
# 打印调用信息
args_repr = [repr(a) for a in args]
kwargs_repr = [f"{k}={v!r}" for k, v in kwargs.items()]
signature = ", ".join(args_repr + kwargs_repr)
print(f"[DEBUG] 调用 {func.__name__}({signature})")
print(f" 位置: {filename}:{lineno}")
# 执行函数
try:
result = func(*args, **kwargs)
print(f"[DEBUG] {func.__name__} 返回: {result!r}")
return result
except Exception as e:
print(f"[DEBUG] {func.__name__} 抛出异常: {e!r}")
raise
return wrapper
# 使用示例
@debug
def calculate(x, y, operation="add"):
if operation == "add":
return x + y
elif operation == "multiply":
return x * y
else:
raise ValueError(f"未知操作: {operation}")
# 测试调试功能
result1 = calculate(10, 5)
result2 = calculate(10, 5, operation="multiply")
🔍 装饰器链追踪
查看函数被哪些装饰器包裹:
from functools import wraps
def trace_decorators(func):
"""显示装饰器链的装饰器"""
if not hasattr(func, '__decorator_chain__'):
func.__decorator_chain__ = []
def decorator(inner_func):
@wraps(inner_func)
def wrapper(*args, **kwargs):
return inner_func(*args, **kwargs)
# 记录装饰器
if not hasattr(wrapper, '__decorator_chain__'):
wrapper.__decorator_chain__ = getattr(inner_func, '__decorator_chain__', [])
wrapper.__decorator_chain__.append(func.__name__)
return wrapper
return decorator
# 创建多个装饰器
@trace_decorators
def logger(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(f"📝 记录: {func.__name__}")
return func(*args, **kwargs)
return wrapper
@trace_decorators
def timer(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(f"⏱️ 计时: {func.__name__}")
return func(*args, **kwargs)
return wrapper
# 使用多个装饰器
@logger
@timer
def my_function():
return "完成"
# 查看装饰器链
print(f"装饰器链: {my_function.__decorator_chain__}")
result = my_function()
⚠️ 常见调试问题
- 函数名丢失:忘记使用
@wraps导致元信息丢失 - 堆栈追踪混乱:多层装饰器使错误信息难以定位
- 循环引用:装饰器互相嵌套导致死循环
- 全局状态:装饰器使用全局变量导致难以测试
实战应用
🔒 示例1:API 访问频率限制
在 Web 应用中,需要限制用户对 API 的访问频率,防止滥用:
import time
from functools import wraps
from collections import defaultdict
class RateLimiter:
"""
访问频率限制装饰器
在指定时间窗口内限制函数调用次数
"""
def __init__(self, max_calls, time_window):
"""
:param max_calls: 最大调用次数
:param time_window: 时间窗口(秒)
"""
self.max_calls = max_calls
self.time_window = time_window
self.calls = defaultdict(list) # 存储每个用户的调用记录
def __call__(self, func):
@wraps(func)
def wrapper(user_id, *args, **kwargs):
now = time.time()
# 清除超出时间窗口的记录
self.calls[user_id] = [
call_time for call_time in self.calls[user_id]
if now - call_time < self.time_window
]
# 检查是否超过频率限制
if len(self.calls[user_id]) >= self.max_calls:
remaining_time = int(self.time_window - (now - self.calls[user_id][0]))
raise Exception(
f"访问频率限制:用户 {user_id} 请在 {remaining_time} 秒后重试"
)
# 记录本次调用
self.calls[user_id].append(now)
return func(user_id, *args, **kwargs)
return wrapper
# 使用示例:限制 10 秒内最多调用 3 次
@RateLimiter(max_calls=3, time_window=10)
def api_endpoint(user_id, data):
"""模拟 API 接口"""
return f"用户 {user_id} 请求成功:{data}"
# 测试限流功能
try:
print(api_endpoint("user_001", "请求1"))
print(api_endpoint("user_001", "请求2"))
print(api_endpoint("user_001", "请求3"))
print(api_endpoint("user_001", "请求4")) # 这个会被拒绝
except Exception as e:
print(f"错误:{e}")
✅ 示例2:参数类型验证装饰器
自动验证函数参数的类型和值,减少样板代码:
from functools import wraps
from typing import get_type_hints
def validate_types(func):
"""
根据类型注解自动验证函数参数类型
"""
@wraps(func)
def wrapper(*args, **kwargs):
# 获取函数的类型注解
hints = get_type_hints(func)
# 验证位置参数
func_args = func.__code__.co_varnames[:func.__code__.co_argcount]
for arg_name, arg_value in zip(func_args, args):
if arg_name in hints:
expected_type = hints[arg_name]
if not isinstance(arg_value, expected_type):
raise TypeError(
f"参数 '{arg_name}' 类型错误:期望 {expected_type.__name__},实际得到 {type(arg_value).__name__}"
)
# 验证关键字参数
for arg_name, arg_value in kwargs.items():
if arg_name in hints:
expected_type = hints[arg_name]
if not isinstance(arg_value, expected_type):
raise TypeError(
f"参数 '{arg_name}' 类型错误:期望 {expected_type.__name__},实际得到 {type(arg_value).__name__}"
)
return func(*args, **kwargs)
return wrapper
# 使用示例
@validate_types
def calculate_total(price: float, quantity: int, discount: float = 0.0) -> float:
"""计算总价(含折扣)"""
return price * quantity * (1 - discount)
# 正确调用
print(calculate_total(99.9, 2, 0.1)) # 179.82
# 错误调用(类型不匹配)
try:
print(calculate_total("99.9", 2)) # 将抛出 TypeError
except TypeError as e:
print(f"错误:{e}")
💾 示例3:数据库事务管理
自动处理数据库事务的提交和回滚:
from functools import wraps
def transactional(func):
"""
数据库事务装饰器
自动处理事务的提交和回滚
"""
@wraps(func)
def wrapper(db_connection, *args, **kwargs):
try:
# 开始事务
print("[事务] 开始")
# 执行业务逻辑
result = func(db_connection, *args, **kwargs)
# 提交事务
db_connection.commit()
print("[事务] 提交成功")
return result
except Exception as e:
# 发生错误,回滚事务
db_connection.rollback()
print(f"[事务] 回滚:{e}")
raise
return wrapper
# 模拟数据库连接
class MockDBConnection:
def commit(self):
print(" -> DB: 执行 COMMIT")
def rollback(self):
print(" -> DB: 执行 ROLLBACK")
def execute(self, sql):
print(f" -> DB: 执行 SQL: {sql}")
# 使用事务装饰器
@transactional
def transfer_money(db, from_account, to_account, amount):
"""转账操作"""
db.execute(f"UPDATE accounts SET balance = balance - {amount} WHERE id = {from_account}")
db.execute(f"UPDATE accounts SET balance = balance + {amount} WHERE id = {to_account}")
return f"从账户 {from_account} 向 {to_account} 转账 {amount} 元"
# 测试事务功能
db = MockDBConnection()
result = transfer_money(db, 1001, 1002, 100)
print(f"结果:{result}")
标准库装饰器
📦 Python 内置装饰器详解
Python 标准库提供了多个常用装饰器,了解它们可以帮助你写出更优雅的代码:
🔑 @property - 属性装饰器
将方法转换为只读属性,实现封装和数据验证:
class Circle:
def __init__(self, radius):
self._radius = radius # 私有属性
@property
def radius(self):
"""获取半径"""
return self._radius
@radius.setter
def radius(self, value):
"""设置半径,带数据验证"""
if value <= 0:
raise ValueError("半径必须大于 0")
self._radius = value
@property
def area(self):
"""计算面积(只读属性)"""
import math
return math.pi * self._radius ** 2
@property
def circumference(self):
"""计算周长(只读属性)"""
import math
return 2 * math.pi * self._radius
# 使用示例
c = Circle(5)
print(f"半径:{c.radius}") # 像访问属性一样
print(f"面积:{c.area:.2f}")
print(f"周长:{c.circumference:.2f}")
c.radius = 10 # 通过 setter 修改
print(f"新面积:{c.area:.2f}")
🏛️ @staticmethod 和 @classmethod
用于定义类级别的方法,不需要实例化:
class DateUtil:
"""日期工具类"""
# 类变量
date_format = "%Y-%m-%d"
@staticmethod
def is_leap_year(year):
"""
静态方法:判断是否为闰年
不需要访问类或实例属性
"""
return (year % 4 == 0 and year % 100 != 0) or (year % 400 == 0)
@classmethod
def from_string(cls, date_string):
"""
类方法:从字符串创建日期对象
可以访问类属性
"""
from datetime import datetime
date_obj = datetime.strptime(date_string, cls.date_format)
return cls(date_obj.year, date_obj.month, date_obj.day)
@classmethod
def today(cls):
"""类方法:获取今天的日期"""
from datetime import datetime
now = datetime.now()
return cls(now.year, now.month, now.day)
def __init__(self, year, month, day):
self.year = year
self.month = month
self.day = day
def __str__(self):
return f"{self.year}-{self.month:02d}-{self.day:02d}"
# 使用静态方法(不需要实例)
print(DateUtil.is_leap_year(2024)) # True
print(DateUtil.is_leap_year(2023)) # False
# 使用类方法创建对象
date1 = DateUtil.from_string("2024-12-25")
print(f"解析的日期:{date1}")
date2 = DateUtil.today()
print(f"今天是:{date2}")
💡 关键区别
@staticmethod:不需要访问类或实例属性,类似于普通函数
@classmethod:第一个参数是类本身(cls),可以访问类属性和创建实例
⚡ @functools.lru_cache - 内置缓存装饰器
Python 标准库提供的高效缓存装饰器,使用 LRU(Least Recently Used)算法:
from functools import lru_cache
import time
# 使用 LRU 缓存,最多缓存 128 个结果
@lru_cache(maxsize=128)
def expensive_computation(n):
"""模拟耗时计算"""
print(f" 计算 {n}...")
time.sleep(1) # 模拟耗时操作
return n * n
print("第一次调用:")
start = time.time()
result1 = expensive_computation(5)
print(f"结果:{result1},耗时:{time.time() - start:.2f}秒\n")
print("第二次调用(缓存命中):")
start = time.time()
result2 = expensive_computation(5)
print(f"结果:{result2},耗时:{time.time() - start:.4f}秒\n")
# 查看缓存统计
print(f"缓存信息:{expensive_computation.cache_info()}")
# 清空缓存
expensive_computation.cache_clear()
print("缓存已清空")
📝 总结
标准库装饰器提供了丰富的功能:
• @property - 属性封装
• @staticmethod / @classmethod - 类方法
• @functools.lru_cache - 结果缓存
• @functools.wraps - 保留元信息
• @dataclasses.dataclass - 数据类
• @contextlib.contextmanager - 上下文管理器