Python装饰器进阶应用：重试机制、性能计时与临时文件处理
基础装饰器原理
无参数装饰器是最基本的形式，通过嵌套函数实现对目标函数的包装。
def timing(func):
def wrapper(*args, **kwargs):
print(f"执行 {func.__name__}")
return func(*args, **kwargs)
return wrapper
@timing
def greet():
return "Hello"
greet()
该模式本质是将原函数替换为包装后的 wrapper 函数，执行流程为：
1. 调用被装饰函数；
2. 实际执行的是装饰器返回的 wrapper；
3. 在调用前/后可插入任意逻辑，如日志记录、权限检查等。
带参数的装饰器设计
当需要动态配置行为时，需使用"装饰器工厂"模式。
def delay(seconds=1):
def decorator(func):
def wrapper(*args, **kwargs):
import time
time.sleep(seconds)
return func(*args, **kwargs)
return wrapper
return decorator
@delay(seconds=2)
def task():
print("任务执行中")
task()
关键点在于：
- 外层函数（如 delay）接收配置参数；
- 返回真正的装饰器函数（如 decorator）；
- 使用时必须加括号调用，否则无法生成有效装饰器。
自动重试机制
用于处理网络请求、文件读写等不稳定操作，避免手动编写循环逻辑。
def retry(max_attempts=3, delay=1):
def decorator(func):
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
result = func(*args, **kwargs)
if result is not False:
return result
except Exception as e:
print(f"第 {attempt + 1} 次失败: {e}")
if attempt < max_attempts - 1:
import time
time.sleep(delay)
return None
return wrapper
return decorator
@retry(max_attempts=5, delay=0.5)
def unreliable_fetch():
import random
if random.random() < 0.7:
raise ConnectionError("模拟连接失败")
return "数据获取成功"
函数执行耗时统计
支持同步与异步函数，可自定义时间单位和精度。
import asyncio
import time
from functools import wraps
def monitor(unit='s', precision=4):
units = {'s': 1, 'ms': 1e3, 'us': 1e6}
unit_labels = {'s': '秒', 'ms': '毫秒', 'us': '微秒'}
def decorator(func):
@wraps(func)
async def async_monitor(*args, **kwargs):
start = time.perf_counter()
result = await func(*args, **kwargs)
duration = (time.perf_counter() - start) * units[unit]
print(f"[{func.__name__}] 耗时: {duration:.{precision}f} {unit_labels[unit]}")
return result
@wraps(func)
def sync_monitor(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
duration = (time.perf_counter() - start) * units[unit]
print(f"[{func.__name__}] 耗时: {duration:.{precision}f} {unit_labels[unit]}")
return result
return async_monitor if asyncio.iscoroutinefunction(func) else sync_monitor
return decorator
@monitor(unit='ms', precision=2)
async def async_task():
await asyncio.sleep(0.1)
return "完成"
@monitor(unit='s', precision=3)
def sync_task():
time.sleep(0.5)
return "完成"
临时文件封装处理
解决路径含中文或特殊字符导致文件读取失败的问题，尤其适用于图像处理库。
import tempfile
import os
def temp_file_wrapper(file_path):
def decorator(func):
def wrapper(*args, **kwargs):
ext = os.path.splitext(file_path)[1]
with tempfile.NamedTemporaryFile(suffix=ext, delete=False, mode='wb') as tmp:
with open(file_path, 'rb') as src:
tmp.write(src.read())
temp_name = tmp.name
try:
# 将临时路径作为第一个参数传入目标函数
return func(temp_name, *args, **kwargs)
finally:
os.unlink(temp_name)
return wrapper
return decorator
# 示例：OpenCV读取中文路径图片
@temp_file_wrapper(r"C:\测试\中文图片.jpg")
def read_image_with_temp(path):
import cv2
img = cv2.imread(path)
return img.shape
# 调用方式一（推荐）
result = read_image_with_temp()
# 调用方式二（显式声明）
def load_image(path):
return temp_file_wrapper(path)(cv2.imread)()
类成员函数的装饰技巧
直接在类内使用装饰器会导致 self 参数错位问题，建议将实例作为参数传入。
def class_logger(instance):
def decorator(func):
def wrapper(*args, **kwargs):
print(f"当前状态: {instance.count}")
instance.count += 1
return func(*args, **kwargs)
return wrapper
return decorator
class Counter:
def __init__(self):
self.count = 0
counter = Counter()
@class_logger(counter)
def process_data(name):
print(f"处理 {name}")
process_data("用户数据")
print(counter.count) # 输出 1
此方法避免了 self 被错误注入装饰器内部，确保上下文正确传递。
