decorator 本质就是一个返回函数的高阶函数,实现 aop
介绍
- functools.wraps 的作用是:保持原来的
__name__
import functools
def log(func):
@functools.wraps(func)
def wrapper(*args, **kw):
print 'call %s():' % func.__name__
return func(*args, **kw)
return wrapper
- 偏函数: functools.partial 的作用就是,把一个函数的某些参数给固定住(也就是设置默认值),返回一个新的函数,调用这个新函数会更简单。当函数的参数个数太多,需要简化时,使用 functools.partial 可以创建一个新的函数,这个新函数可以固定住原函数的部分参数,从而在调用时更简单
示例
计算方法耗时装饰器
import time
# (复用上面的 time_it 装饰器)
def time_it(func):
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
elapsed_time = (end_time - start_time) * 1000
print(f"方法 '{func.__name__}' 执行耗时: {elapsed_time:.2f} ms")
return result
return wrapper
class DataProcessor:
def __init__(self, name):
self.name = name
@time_it
def process_data(self):
"""处理数据的实例方法"""
print(f"'{self.name}' 开始处理数据...")
time.sleep(1.5)
print("数据处理完成。")
@staticmethod
@time_it
def utility_function():
"""一个静态工具方法"""
print("执行静态工具方法...")
time.sleep(0.5)
print("静态方法执行完毕。")
# --- 使用 ---
processor = DataProcessor("处理器A")
processor.process_data()
print("-" * 20)
DataProcessor.utility_function()
import time
import asyncio
from functools import wraps
def async_time_it(func):
"""一个用于异步函数的计时装饰器"""
@wraps(func)
# 1. wrapper 必须是 async def
async def wrapper(*args, **kwargs):
start_time = time.time()
# 2. 必须使用 await 来执行原始的异步函数
result = await func(*args, **kwargs)
end_time = time.time()
elapsed_time = (end_time - start_time) * 1000
print(f"异步方法 '{func.__name__}' 执行耗时: {elapsed_time:.2f} ms")
return result
return wrapper
# --- 使用装饰器 ---
@async_time_it
async def async_slow_function(delay_seconds):
"""一个模拟IO阻塞的异步函数"""
print(f"协程开始执行,预计等待 {delay_seconds} 秒...")
await asyncio.sleep(delay_seconds)
print("协程执行完毕。")
return "Async OK"
# --- 调用被装饰的异步函数 ---
async def main():
print("--- 调用 async_slow_function ---")
status = await async_slow_function(2)
print(f"async_slow_function 返回结果: {status}\n")
# 在 Python 3.7+ 中,可以直接运行
if __name__ == "__main__":
asyncio.run(main())
import time
import asyncio
from functools import wraps
def universal_time_it_robust(func):
@wraps(func)
async def async_wrapper(*args, **kwargs):
start_time = time.time()
result = await func(*args, **kwargs)
end_time = time.time()
elapsed_time = (end_time - start_time) * 1000
print(f"方法(async) '{func.__name__}' 执行耗时: {elapsed_time:.2f} ms")
return result
@wraps(func)
def sync_wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
elapsed_time = (end_time - start_time) * 1000
print(f"方法(sync) '{func.__name__}' 执行耗时: {elapsed_time:.2f} ms")
return result
# 在装饰时就决定返回哪个 wrapper
if asyncio.iscoroutinefunction(func):
return async_wrapper
else:
return sync_wrapper
# --- 使用通用装饰器 ---
@universal_time_it_robust
def sync_task():
time.sleep(1)
@universal_time_it_robust
async def async_task():
await asyncio.sleep(1.5)
async def main_universal():
print("--- 执行同步任务 ---")
sync_task()
print("\n--- 执行异步任务 ---")
await async_task()
if __name__ == "__main__":
asyncio.run(main_universal())