你是否曾遇到过这样的困扰：代码中重复的日志逻辑散落在多个函数中，相同的缓存代码被复制粘贴了无数次，每次添加新功能都要手动插入权限检查？如果你的答案是肯定的，那么Python装饰器正是你需要的解决方案。
这个看似简单的语法糖，实则是一把改变编码方式的“瑞士军刀”，它能将混乱的“意大利面条式”代码重构为整洁的“乐高积木”。本文将深入解析10个高级Python装饰器模式，每一个都针对实际开发中的痛点，助你彻底告别重复代码。
为什么装饰器如此强大？
在深入代码之前，我们先理解装饰器的本质。想象一下，你需要为房屋的每个房间安装相同的智能照明系统。笨方法是在每个房间手动铺设电路；聪明方法则是在建造时就将系统作为标准配置集成。装饰器正是那个“聪明方法”。
装饰器的核心价值在于分离关注点。业务逻辑专注于业务，而横切关注点（如日志、缓存、权限）则通过装饰器独立处理，互不干扰。理解了这一理念，让我们开始今天的装饰器探索之旅。
第一部分：性能优化类装饰器
1. TTL缓存装饰器：让缓存自动过期
应用场景：使用 functools.lru_cache 时，某些数据会随时间变化，需要定期刷新。
痛点：标准LRU缓存不会过期，可能导致返回陈旧数据。
解决方案：TTL（Time-To-Live）缓存装饰器。
import functools
import time
def ttl_cache(ttl_seconds):
"""缓存装饰器，支持自动过期"""
cache = {}
def decorator(func):
@functools.wraps(func)
def wrapper(*args):
now = time.time()
# 检查缓存是否存在且未过期
if args in cache:
value, timestamp = cache[args]
if now - timestamp < ttl_seconds:
return value
# 缓存未命中或已过期，重新计算
result = func(*args)
cache[args] = (result, now)
return result
return wrapper
return decorator
# 使用示例
@ttl_cache(ttl_seconds=3)
def get_weather(city):
"""模拟获取天气数据的昂贵操作"""
print(f"Fetching weather for {city}...")
time.sleep(0.5) # 模拟网络延迟
return f"Weather in {city}: Sunny, 25°C"
# 测试
print("第一次调用（计算并缓存）:")
print(get_weather("Beijing"))
print("\n第二次调用（1秒后，从缓存读取）:")
time.sleep(1)
print(get_weather("Beijing"))
print("\n第三次调用（4秒后，缓存过期，重新计算）:")
time.sleep(4)
print(get_weather("Beijing"))
输出效果：
第一次调用（计算并缓存）:
Fetching weather for Beijing...
Weather in Beijing: Sunny, 25°C
第二次调用（1秒后，从缓存读取）:
Weather in Beijing: Sunny, 25°C
第三次调用（4秒后，缓存过期，重新计算）:
Fetching weather for Beijing...
Weather in Beijing: Sunny, 25°C
关键点：
functools.wraps 确保装饰后的函数保留原始函数的元数据- 使用字典存储
(结果, 时间戳) 对
- 参数
args 作为缓存键（对于带有关键字参数的函数需要更复杂的处理）
2. 执行时间跟踪器：快速定位性能瓶颈
应用场景：应用性能下降，但无法确定具体是哪个函数拖慢了速度。
痛点：手动添加时间测量代码既繁琐又容易遗漏。
解决方案：通用执行时间跟踪装饰器。
import time
import functools
def timeit(func):
"""测量函数执行时间的装饰器"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter() # 使用高精度计时器
result = func(*args, **kwargs)
elapsed = (time.perf_counter() - start) * 1000 # 转换为毫秒
# 在实际项目中，这里应该使用日志系统而非print
print(f"[PERF] {func.__name__} executed in {elapsed:.2f} ms")
return result
return wrapper
@timeit
def process_large_data(n):
"""处理大量数据的模拟函数"""
return sum(i * i for i in range(n))
@timeit
def fetch_from_database(query):
"""模拟数据库查询"""
time.sleep(0.1)
return f"Results for {query}"
# 测试性能
process_large_data(1_000_000)
fetch_from_database("SELECT * FROM users")
输出：
[PERF] process_large_data executed in 46.32 ms
[PERF] fetch_from_database executed in 100.15 ms
进阶技巧：在实际项目中，可以将这个装饰器与日志级别结合，只在需要性能分析时启用它。
第二部分：系统健壮性类装饰器
3. 失败重试装饰器：应对不稳定操作
应用场景：网络请求、文件I/O、第三方API调用等可能偶尔失败的操作。
痛点：到处写重试逻辑使代码变得臃肿。
解决方案：可配置的重试装饰器。
import functools
import time
import random
def retry(retries=3, delay=1.0, backoff=2.0, exceptions=(Exception,)):
"""
重试装饰器
参数:
retries: 最大重试次数
delay: 首次重试延迟（秒）
backoff: 延迟倍数（用于指数退避）
exceptions: 触发重试的异常类型
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
current_delay = delay
for attempt in range(1, retries + 1):
try:
return func(*args, **kwargs)
except exceptions as e:
if attempt == retries:
print(f"All {retries} attempts failed. Last error: {e}")
raise
print(f"Attempt {attempt} failed: {e}. Retrying in {current_delay:.1f}s...")
time.sleep(current_delay)
current_delay *= backoff # 指数退避
# 理论上不会执行到这里
return None
return wrapper
return decorator
# 模拟一个不稳定的API调用
@retry(retries=4, delay=0.5, backoff=1.5, exceptions=(ConnectionError, TimeoutError))
def call_unstable_api(endpoint):
"""模拟不稳定的API调用"""
if random.random() < 0.7: # 70%概率失败
raise ConnectionError(f"Failed to connect to {endpoint}")
return f"Success from {endpoint}"
# 测试
result = call_unstable_api("/api/data")
print(f"最终结果: {result}")
输出示例：
Attempt 1 failed: Failed to connect to /api/data. Retrying in 0.5s...
Attempt 2 failed: Failed to connect to /api/data. Retrying in 0.8s...
Attempt 3 failed: Failed to connect to /api/data. Retrying in 1.2s...
最终结果: Success from /api/data
指数退避策略：这是处理网络问题的黄金标准。每次失败后等待时间增加，避免给已经过载的服务增加压力。
4. 限流装饰器：防止API滥用
应用场景：公开API、用户操作限制、防止爬虫等。
痛点：在业务逻辑中嵌入限流检查。
解决方案：基于时间的限流装饰器。
import time
import functools
def rate_limit(calls_per_minute):
"""限制每分钟调用次数的装饰器"""
interval = 60.0 / calls_per_minute
def decorator(func):
last_called = [0.0] # 使用列表实现闭包内的可变状态
@functools.wraps(func)
def wrapper(*args, **kwargs):
elapsed = time.time() - last_called[0]
if elapsed < interval:
# 调用太频繁
wait_time = interval - elapsed
print(f"Rate limit exceeded. Please wait {wait_time:.1f} seconds.")
return None
last_called[0] = time.time()
return func(*args, **kwargs)
return wrapper
return decorator
# 限制为每分钟最多10次调用
@rate_limit(calls_per_minute=10)
def send_notification(user_id, message):
"""发送通知（模拟）"""
print(f"Notification sent to user {user_id}: {message}")
return True
# 测试快速连续调用
for i in range(15):
result = send_notification(123, f"Message {i}")
if result is None:
time.sleep(0.1) # 稍微等待后继续测试
线程安全版：对于多线程环境，需要使用线程锁：
import threading
def rate_limit_threadsafe(calls_per_minute):
interval = 60.0 / calls_per_minute
lock = threading.Lock()
last_called = 0.0
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
nonlocal last_called
with lock:
now = time.time()
elapsed = now - last_called
if elapsed < interval:
return None
last_called = now
return func(*args, **kwargs)
return wrapper
return decorator
第三部分：代码可维护性类装饰器
5. 简易日志装饰器：告别调试print语句
应用场景：调试时临时添加的print语句经常忘记删除，导致生产环境日志混乱。
痛点：日志逻辑与业务逻辑混杂。
解决方案：非侵入式日志装饰器。
import functools
import datetime
def log_call(level="INFO"):
"""记录函数调用的装饰器"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
# 记录调用信息
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
arg_str = ", ".join([repr(arg) for arg in args])
kwarg_str = ", ".join([f"{k}={repr(v)}" for k, v in kwargs.items()])
all_args = ", ".join(filter(None, [arg_str, kwarg_str]))
print(f"[{timestamp}] [{level}] Calling {func.__name__}({all_args})")
# 执行函数
result = func(*args, **kwargs)
# 记录返回结果
print(f"[{timestamp}] [{level}] {func.__name__} returned {repr(result)}")
return result
return wrapper
return decorator
@log_call(level="DEBUG")
def calculate_discount(price, discount_rate=0.1, tax_rate=0.08):
"""计算折扣后价格"""
discounted = price * (1 - discount_rate)
final_price = discounted * (1 + tax_rate)
return round(final_price, 2)
# 测试
price = calculate_discount(100, discount_rate=0.2)
print(f"最终价格: ${price}")
输出：
[2024-01-15 10:30:00] [DEBUG] Calling calculate_discount(100, discount_rate=0.2)
[2024-01-15 10:30:00] [DEBUG] calculate_discount returned 86.4
最终价格: $86.4
生产环境建议：在实际项目中，应该使用Python的 logging 模块替代 print，并支持日志级别过滤。
6. 依赖注入装饰器：告别全局变量
应用场景：函数需要访问数据库连接、配置、日志器等共享资源。
痛点：使用全局变量或到处传递依赖。
解决方案：依赖注入装饰器。
import functools
def inject(**dependencies):
"""依赖注入装饰器"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
# 临时将依赖注入到函数的全局命名空间
original_globals = func.__globals__.copy()
try:
# 添加依赖
func.__globals__.update(dependencies)
return func(*args, **kwargs)
finally:
# 恢复原始全局变量
for key in dependencies:
func.__globals__.pop(key, None)
# 对于已存在的键，恢复其原始值
for key, value in original_globals.items():
if key not in func.__globals__:
func.__globals__[key] = value
return wrapper
return decorator
# 模拟一些依赖项
class Database:
def query(self, sql):
return f"Result: {sql}"
class Logger:
def info(self, msg):
print(f"[INFO] {msg}")
# 创建依赖实例
db = Database()
logger = Logger()
# 使用依赖注入
@inject(db=db, logger=logger)
def process_order(order_id):
logger.info(f"Processing order {order_id}")
result = db.query(f"SELECT * FROM orders WHERE id={order_id}")
logger.info(f"Query result: {result}")
return result
# 测试
process_order(12345)
更安全的方法：上面的方法修改了全局命名空间，可能带来副作用。更安全的方法是显式传递依赖：
def inject_safe(**dependencies):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
# 将依赖作为关键字参数传递
return func(*args, **dependencies, **kwargs)
return wrapper
return decorator
7. 类装饰器：一次性装饰所有方法
应用场景：你需要给一个类的所有方法添加相同的行为（如日志、计时、权限检查）。
痛点：给每个方法单独添加装饰器。
解决方案：类级别装饰器。
import types
import functools
import time
def time_all_methods(cls):
"""装饰类中所有方法的装饰器"""
# 遍历类的所有属性
for attr_name in dir(cls):
# 跳过特殊方法和私有方法
if attr_name.startswith("_"):
continue
attr = getattr(cls, attr_name)
# 只装饰可调用方法
if callable(attr):
# 创建装饰后的方法
@functools.wraps(attr)
def timed_method(self, *args, __original_method=attr, **kwargs):
start = time.perf_counter()
result = __original_method(self, *args, **kwargs)
elapsed = (time.perf_counter() - start) * 1000
print(f"{cls.__name__}.{__original_method.__name__} took {elapsed:.2f} ms")
return result
# 替换原始方法
setattr(cls, attr_name, timed_method)
return cls
@time_all_methods
class DataProcessor:
def __init__(self, data):
self.data = data
def filter_data(self):
"""过滤数据"""
time.sleep(0.05)
return [x for x in self.data if x % 2 == 0]
def transform_data(self):
"""转换数据"""
time.sleep(0.03)
return [x * 2 for x in self.data]
def process(self):
"""完整处理流程"""
filtered = self.filter_data()
transformed = self.transform_data()
return transformed
# 测试
processor = DataProcessor(list(range(1000)))
result = processor.process()
print(f"处理结果长度: {len(result)}")
输出：
DataProcessor.filter_data took 50.12 ms
DataProcessor.transform_data took 30.05 ms
DataProcessor.process took 80.25 ms
处理结果长度: 1000
注意事项：这种装饰器会影响类的所有实例。如果你只需要装饰特定方法，建议使用元类或显式装饰。
第四部分：高级架构类装饰器
8. 单例装饰器：确保全局唯一实例
应用场景：配置管理、数据库连接池、缓存管理器等需要全局唯一实例的类。
痛点：手动实现单例模式需要重复代码。
解决方案：通用单例装饰器。
import functools
def singleton(cls):
"""单例装饰器"""
instances = {}
@functools.wraps(cls)
def wrapper(*args, **kwargs):
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return wrapper
@singleton
class AppConfig:
def __init__(self):
print("Initializing AppConfig...")
self.settings = {
"debug": True,
"database_url": "postgresql://localhost/mydb",
"cache_ttl": 300
}
def get(self, key):
return self.settings.get(key)
def set(self, key, value):
self.settings[key] = value
# 测试单例行为
config1 = AppConfig()
config2 = AppConfig()
print(f"config1 is config2: {config1 is config2}")
print(f"config1 ID: {id(config1)}")
print(f"config2 ID: {id(config2)}")
# 验证配置共享
config1.set("debug", False)
print(f"config2.get('debug'): {config2.get('debug')}")
输出：
Initializing AppConfig...
config1 is config2: True
config1 ID: 140000000000000
config2 ID: 140000000000000
config2.get('debug'): False
线程安全版本：对于多线程环境：
import threading
def singleton_threadsafe(cls):
instances = {}
lock = threading.Lock()
@functools.wraps(cls)
def wrapper(*args, **kwargs):
if cls not in instances:
with lock:
if cls not in instances: # 双重检查锁定
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return wrapper
9. 基于角色的访问控制装饰器
应用场景：Web应用、API服务、管理后台等需要权限控制的系统。
痛点：在每个需要权限检查的函数中重复验证逻辑。
解决方案：RBAC（基于角色的访问控制）装饰器。
import functools
def require_role(*allowed_roles):
"""基于角色的权限检查装饰器"""
def decorator(func):
@functools.wraps(func)
def wrapper(user, *args, **kwargs):
user_role = user.get("role", "guest")
if user_role not in allowed_roles:
raise PermissionError(
f"User {user.get('name', 'Unknown')} with role '{user_role}' "
f"is not allowed to perform {func.__name__}. "
f"Required roles: {allowed_roles}"
)
return func(user, *args, **kwargs)
return wrapper
return decorator
# 模拟用户数据
admin_user = {"id": 1, "name": "Alice", "role": "admin"}
editor_user = {"id": 2, "name": "Bob", "role": "editor"}
viewer_user = {"id": 3, "name": "Charlie", "role": "viewer"}
class ContentManagementSystem:
@require_role("admin", "editor")
def create_article(self, user, title, content):
print(f"{user['name']} created article: {title}")
return {"id": 100, "title": title, "content": content}
@require_role("admin", "editor", "viewer")
def view_article(self, user, article_id):
print(f"{user['name']} viewed article {article_id}")
return {"id": article_id, "content": "Sample content"}
@require_role("admin")
def delete_article(self, user, article_id):
print(f"{user['name']} deleted article {article_id}")
return {"success": True}
# 测试
cms = ContentManagementSystem()
# 正常情况
cms.create_article(admin_user, "Python Tips", "Some content")
cms.view_article(viewer_user, 100)
# 权限不足的情况
try:
cms.delete_article(editor_user, 100)
except PermissionError as e:
print(f"权限错误: {e}")
输出：
Alice created article: Python Tips
Charlie viewed article 100
权限错误: User Bob with role 'editor' is not allowed to perform delete_article. Required roles: ('admin',)
10. 上下文管理装饰器：传递请求级信息
应用场景：微服务、异步任务、Web请求处理等需要传递上下文信息的场景。
痛点：通过函数参数层层传递上下文信息（如请求ID、用户信息）。
解决方案：基于 contextvars 的上下文管理装饰器。
import functools
import contextvars
import uuid
# 创建上下文变量
request_id_var = contextvars.ContextVar("request_id", default=None)
user_context_var = contextvars.ContextVar("user", default=None)
def with_context(func):
"""上下文管理装饰器"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
# 获取当前上下文
request_id = request_id_var.get()
user = user_context_var.get()
# 如果没有请求ID，生成一个
if request_id is None:
request_id = str(uuid.uuid4())[:8]
request_id_var.set(request_id)
print(f"[{request_id}] Starting {func.__name__}")
if user:
print(f"[{request_id}] User: {user.get('name', 'Unknown')}")
try:
result = func(*args, **kwargs)
print(f"[{request_id}] {func.__name__} completed successfully")
return result
except Exception as e:
print(f"[{request_id}] {func.__name__} failed: {e}")
raise
return wrapper
def set_request_context(request_id=None, user=None):
"""设置请求上下文"""
if request_id:
request_id_var.set(request_id)
if user:
user_context_var.set(user)
# 模拟一个Web请求处理流程
@with_context
def authenticate(token):
print(f"Authenticating token: {token[:10]}...")
return {"id": 123, "name": "John Doe", "role": "user"}
@with_context
def process_request(data):
user = user_context_var.get()
print(f"Processing request for user: {user['name']}")
return {"status": "success", "data": data}
@with_context
def handle_api_request(token, request_data):
# 认证
user = authenticate(token)
user_context_var.set(user)
# 处理请求
result = process_request(request_data)
# 记录日志
request_id = request_id_var.get()
print(f"[{request_id}] API request completed")
return result
# 测试
print("=== 测试1：正常请求 ===")
set_request_context(request_id="req_001")
response = handle_api_request("secure_token_123", {"action": "get_data"})
print(f"Response: {response}")
print("\n=== 测试2：另一个请求 ===")
set_request_context(request_id="req_002", user={"id": 456, "name": "Jane Smith"})
response = handle_api_request("secure_token_456", {"action": "update_data"})
print(f"Response: {response}")
输出：
=== 测试1：正常请求 ===
[req_001] Starting handle_api_request
Authenticating token: secure_tok...
[req_001] Starting authenticate
[req_001] authenticate completed successfully
[req_001] Starting process_request
Processing request for user: John Doe
[req_001] process_request completed successfully
[req_001] API request completed
Response: {'status': 'success', 'data': {'action': 'get_data'}}
=== 测试2：另一个请求 ===
[req_002] Starting handle_api_request
[req_002] User: Jane Smith
Authenticating token: secure_tok...
[req_002] Starting authenticate
[req_002] authenticate completed successfully
[req_002] Starting process_request
Processing request for user: John Doe
[req_002] process_request completed successfully
[req_002] API request completed
Response: {'status': 'success', 'data': {'action': 'update_data'}}
异步支持：contextvars 是Python 3.7+引入的特性，完美支持异步操作。上下文变量会自动在异步任务间传播，这是传统的线程局部变量（threading.local）无法做到的。
写在最后
通过以上10个高级Python装饰器模式的探讨，我们从性能优化延伸到系统架构，每一个模式都直击开发中的常见痛点。装饰器的核心价值在于让横切关注点与业务逻辑分离，这种分离带来多重好处：
- 代码更干净：业务逻辑不再被各种辅助代码污染
- 维护更容易：功能修改只需在一个地方进行
- 复用性更强：装饰器可以在不同项目、不同函数间共享
- 可读性更高：通过装饰器名称就能理解函数的附加行为
在软件设计模式的应用中，装饰器模式展现了其强大的灵活性。然而，装饰器虽好，但切勿过度使用。当一个函数被多层装饰器包裹时，调试会变得困难。建议保持装饰器简单透明，使用 functools.wraps 保留元数据，并为复杂装饰器编写单元测试。
希望这些模式能帮助你提升代码质量。更多关于Python和软件架构的深度内容，欢迎在云栈社区交流探讨。
发表于 4 天前
|
查看: 10|
回复: 0
