Python decorator 注解

发布时间: 更新时间: 总字数:968 阅读时间:2m 作者: IP上海 分享 网址

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())
本文总阅读量 次 本站总访问量 次 本站总访客数
Home Archives Categories Tags Statistics