browser-use 库中异步函数执行时间装饰器

代码

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_time
			logger.debug(f'{additional_text} Execution time: {execution_time:.2f} seconds')
			return result

		return wrapper

	return 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 result
    return 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 processed

asyncio.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