2345浏览器下载_网页设计结课报告_厦门站长优化工具_搜索引擎优化营销

代码

import asyncio
import logging
import time
from functools import wraps
from typing import Any, Callable, Coroutine, ParamSpec, TypeVar# 设置日志
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)# 定义类型变量
R = TypeVar('R')
P = ParamSpec('P')def time_execution_async(additional_text: str = '',
) -> Callable[[Callable[P, Coroutine[Any, Any, R]]], Callable[P, Coroutine[Any, Any, R]]]:def decorator(func: Callable[P, Coroutine[Any, Any, R]]) -> Callable[P, Coroutine[Any, Any, R]]:@wraps(func)async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:start_time = time.time()result = await func(*args, **kwargs)execution_time = time.time() - start_timelogger.debug(f'{additional_text} Execution time: {execution_time:.2f} seconds')return resultreturn wrapperreturn decorator

代码解释

这段代码实现了一个用于测量异步函数执行时间的装饰器。

1. 导入和配置：

import asyncio, logging, time
from functools import wraps
from typing import Any, Callable, Coroutine, ParamSpec, TypeVar# 配置日志级别为DEBUG以显示执行时间
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

2. 类型变量定义：

R = TypeVar('R')  # 用于表示函数返回值类型
P = ParamSpec('P')  # 用于表示函数参数类型

3. 装饰器实现：

def time_execution_async(additional_text: str = ''):

• 外层函数：接收装饰器参数
• 参数 additional_text：用于在日志中标识不同的执行时间记录

4. 装饰器核心结构：

def decorator(func):      # 接收被装饰的函数@wraps(func)         # 保持原函数的元数据async def wrapper(*args, **kwargs):  # 包装函数start_time = time.time()         # 记录开始时间result = await func(*args, **kwargs)  # 执行原函数execution_time = time.time() - start_time  # 计算执行时间logger.debug(f'{additional_text} Execution time: {execution_time:.2f} seconds')return resultreturn wrapper
return decorator

主要功能：

1. 测量异步函数执行时间
2. 保持原函数的类型信息和签名
3. 通过日志输出执行时间
4. 支持自定义日志标识文本

使用场景：

• 性能分析
• 执行时间监控
• 异步操作调试
• 性能瓶颈定位

示例

# 示例1：基本使用
@time_execution_async('--fetch-data')
async def fetch_data():await asyncio.sleep(1)  # 模拟网络请求return {'data': 'example'}# 示例2：带参数的函数
@time_execution_async('--process-user')
async def process_user(user_id: int, update: bool = False):await asyncio.sleep(0.5)  # 模拟处理时间return f"User {user_id} processed"# 使用示例
async def main():# 调用装饰后的函数data = await fetch_data()result = await process_user(123, update=True)print(data)  # {'data': 'example'}print(result)  # User 123 processedasyncio.run(main())

python test_record.py
DEBUG:asyncio:Using proactor: IocpProactor
DEBUG:__main__:--fetch-data Execution time: 1.00 seconds
DEBUG:__main__:--process-user Execution time: 0.50 seconds
{'data': 'example'}
User 123 processed