Python装饰器实战：从零到精通的5个经典场景
Python装饰器（Decorator）是一个非常强大且优雅的语言特性，它允许我们在不修改原函数代码的情况下，为函数添加额外的功能。本文将通过5个实战场景，带你深入理解装饰器的原理和应用。
一、装饰器的工作原理
装饰器本质上是一个高阶函数，它接受一个函数作为参数，并返回一个新的函数。在Python中，我们使用@符号来应用装饰器，这其实是"函数调用"的语法糖。
import time
# 一个简单的装饰器示例
def timing_decorator(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
# 使用装饰器
@timing_decorator
def calculate_fibonacci(n):
"""计算斐波那契数列"""
if n <= 1:
return n
return calculate_fibonacci(n-1) calculate_fibonacci(n-2)
# 调用函数
calculate_fibonacci(10)
这个装饰器会自动记录被装饰函数的执行时间。当我们调用calculate_fibonacci(10)时，实际上是调用了wrapper函数，它在原函数前后添加了计时逻辑。
二、场景1：日志记录装饰器
在实际开发中，我们经常需要记录函数的调用信息。一个灵活的日志装饰器可以大大简化这一工作。
import logging
from functools import wraps
# 配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
def log_decorator(log_level=logging.INFO):
"""可配置的日志装饰器"""
def decorator(func):
@wraps(func) # 保留原函数的元数据
def wrapper(*args, **kwargs):
logger.log(log_level, f"调用函数: {func.__name__}")
logger.log(log_level, f"参数: args={args}, kwargs={kwargs}")
try:
result = func(*args, **kwargs)
logger.log(log_level, f"返回值: {result}")
return result
except Exception as e:
logger.error(f"函数 {func.__name__} 执行出错: {str(e)}")
raise
return wrapper
return decorator
# 应用日志装饰器
@log_decorator(log_level=logging.INFO)
def divide_numbers(a, b):
"""除法函数"""
return a / b
# 测试
divide_numbers(10, 2)
divide_numbers(10, 0) # 会触发异常
这个日志装饰器使用了functools.wraps来保留原函数的元数据（如__name__、____doc__等），并且可以配置日志级别。它能记录函数调用、参数、返回值和异常信息。
三、场景2：缓存装饰器
对于计算密集型的函数，我们可以使用缓存来避免重复计算，显著提升性能。
from functools import lru_cache
import time
# Python内置的缓存装饰器
@lru_cache(maxsize=128)
def expensive_calculation(x):
"""模拟耗时计算"""
time.sleep(0.1) # 模拟计算耗时
return x * x
# 自定义缓存装饰器（支持更灵活的过期策略）
class CacheDecorator:
"""带过期时间的缓存装饰器"""
def __init__(self, ttl=60):
self.cache = {}
self.ttl = ttl # 缓存过期时间（秒）
def __call__(self, func):
@wraps(func)
def wrapper(*args, **kwargs):
# 生成缓存键
cache_key = (args, frozenset(kwargs.items()))
# 检查缓存
if cache_key in self.cache:
result, timestamp = self.cache[cache_key]
if time.time() - timestamp < self.ttl:
print("从缓存获取结果")
return result
# 计算新结果
result = func(*args, **kwargs)
self.cache[cache_key] = (result, time.time())
print("计算并缓存结果")
return result
return wrapper
@CacheDecorator(ttl=30)
def get_user_data(user_id):
"""获取用户数据（模拟API调用）"""
time.sleep(0.2) # 模拟网络请求
return {"id": user_id, "name": f"User{user_id}", "email": f"user{user_id}@example.com"}
# 测试缓存
print(get_user_data(1)) # 首次计算
print(get_user_data(1)) # 从缓存获取
print(get_user_data(2)) # 再次计算
Python的functools.lru_cache提供了现成的缓存功能，而自定义的CacheDecorator可以支持更复杂的缓存策略，比如过期时间、缓存大小限制等。
四、场景3：权限验证装饰器
在Web应用中，权限验证是必不可少的功能。装饰器可以让权限验证逻辑与业务逻辑解耦。
from functools import wraps
# 模拟用户数据库
users = {
1: {"name": "管理员", "role": "admin"},
2: {"name": "编辑", "role": "editor"},
3: {"name": "普通用户", "role": "user"}
}
def require_role(required_role):
"""权限验证装饰器"""
def decorator(func):
@wraps(func)
def wrapper(user_id, *args, **kwargs):
# 获取用户信息
user = users.get(user_id)
if not user:
raise PermissionError(f"用户ID {user_id} 不存在")
# 检查权限
user_role = user["role"]
# 管理员拥有所有权限
if user_role == "admin":
pass
# 编辑可以访问编辑和用户权限
elif required_role == "editor" and user_role not in ["admin", "editor"]:
raise PermissionError(f"权限不足: 需要 {required_role} 权限")
# 普通用户权限
elif required_role == "user" and user_role != "user":
raise PermissionError(f"权限不足: 需要 {required_role} 权限")
else:
if user_role != required_role and user_role != "admin":
raise PermissionError(f"权限不足: 需要 {required_role} 权限")
return func(user_id, *args, **kwargs)
return wrapper
return decorator
@require_role("admin")
def delete_post(user_id, post_id):
"""删除文章（管理员权限）"""
print(f"用户 {user_id} 删除了文章 {post_id}")
return True
@require_role("editor")
def edit_post(user_id, post_id, content):
"""编辑文章（编辑权限）"""
print(f"用户 {user_id} 编辑了文章 {post_id}")
return True
@require_role("user")
def view_post(user_id, post_id):
"""查看文章（用户权限）"""
print(f"用户 {user_id} 查看了文章 {post_id}")
return True
# 测试权限验证
try:
delete_post(1, 100) # 管理员可以删除
edit_post(2, 100, "新内容") # 编辑可以编辑
view_post(3, 100) # 用户可以查看
delete_post(3, 100) # 用户不能删除，会抛出异常
except PermissionError as e:
print(f"错误: {e}")
这个权限验证装饰器使用了闭包技术，可以接受参数并生成不同的验证逻辑。它将权限验证与业务逻辑完全分离，代码更加清晰和可维护。
五、场景4：重试机制装饰器
在调用外部API或执行可能失败的操作时，重试机制是一个常见的需求。
import time
from functools import wraps
def retry(max_attempts=3, delay=1, backoff=2):
"""重试装饰器
Args:
max_attempts: 最大重试次数
delay: 初始延迟时间（秒）
backoff: 退避因子
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
current_delay = delay
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
print(f"尝试 {attempt 1}/{max_attempts} 失败: {str(e)}")
if attempt < max_attempts - 1:
print(f"等待 {current_delay} 秒后重试...")
time.sleep(current_delay)
current_delay *= backoff
raise last_exception
return wrapper
return decorator
@retry(max_attempts=3, delay=1, backoff=2)
def fetch_api_data(url):
"""模拟API请求（可能失败）"""
import random
if random.random() < 0.7: # 70%概率失败
raise ConnectionError("网络连接失败")
return {"status": "success", "data": [1, 2, 3]}
# 测试重试机制
try:
result = fetch_api_data("https://api.example.com/data")
print(f"成功获取数据: {result}")
except Exception as e:
print(f"重试次数用尽，最终失败: {e}")
这个重试装饰器支持指数退避策略，每次重试的延迟时间会逐渐增加。这对于处理暂时性的网络故障或服务不可用情况非常有用。
六、场景5：性能监控装饰器
在生产环境中，监控函数的性能指标对于系统优化非常重要。
import time
from functools import wraps
from collections import defaultdict
class PerformanceMonitor:
"""性能监控器"""
def __init__(self):
self.stats = defaultdict(lambda: {"count": 0, "total_time": 0, "min_time": float("inf"), "max_time": 0})
def record(self, func_name, execution_time):
"""记录性能数据"""
self.stats[func_name]["count"] = 1
self.stats[func_name]["total_time"] = execution_time
self.stats[func_name]["min_time"] = min(self.stats[func_name]["min_time"], execution_time)
self.stats[func_name]["max_time"] = max(self.stats[func_name]["max_time"], execution_time)
def report(self):
"""生成性能报告"""
report = []
for func_name, data in self.stats.items():
avg_time = data["total_time"] / data["count"]
report.append(f"{func_name}: 调用{data['count']}次, 平均{avg_time:.4f}秒, 最小{data['min_time']:.4fari}秒, 最大{data['max_time']:.4f}秒")
return "\n".join(report)
# 创建监控器实例
monitor = PerformanceMonitor()
def performance_monitor(func):
"""性能监控装饰器"""
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
execution_time = time.time() - start_time
monitor.record(func.__name__, execution_time)
return result
return wrapper
@performance_monitor
def process_data(size):
"""处理数据（模拟计算）"""
time.sleep(size * 0.01)
return size * 2
@performance_monitor
def save_data(data):
"""保存数据（模拟IO操作）"""
time.sleep(len(data) * 0.005)
return True
# 测试性能监控
for i in range(10):
process_data(i 1)
save_data([1, 2, 3, 4, 5])
print("性能报告:")
print(monitor.report())
这个性能监控装饰器会记录每个函数的调用次数、总执行时间、最小/最大执行时间，并可以生成详细的性能报告。这对于识别性能瓶颈和优化代码非常有帮助。
七、最佳实践和注意事项
在使用装饰器时，有几个重要的最佳实践需要牢记：
1. 使用functools.wraps：总是使用@wraps装饰器来保留原函数的元数据，这对调试和文档生成非常重要。
2. 保持装饰器简单：装饰器应该专注于单一职责，不要在装饰器中添加太多复杂的逻辑。
3. 考虑性能影响：装饰器会增加函数调用的开销，在性能敏感的场景下要谨慎使用。
4. 文档化装饰器：为装饰器编写清晰的文档，说明其用途、参数和效果。
5. 测试装饰器：装饰器也需要单元测试，确保在各种边界情况下都能正常工作。
八、总结
Python装饰器是一个非常强大的工具，它可以让我们的代码更加优雅、模块化和可维护。通过本文的5个实战场景，我们看到了装饰器在实际开发中的广泛应用：
• 日志记录：自动记录函数调用信息
• 缓存优化：避免重复计算，提升性能
• 权限验证：将验证逻辑与业务逻辑解耦
• 重试机制：处理暂时性故障
• 性能监控：收集性能指标，优化系统
掌握装饰器不仅能提升代码质量，还能让你写出更加Pythonic的代码。希望本文的实战示例能帮助你更好地理解和应用这一强大的语言特性。
