mirror of
https://github.com/browser-use/browser-use
synced 2026-05-06 17:52:15 +02:00
60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
import logging
|
|
import time
|
|
from functools import wraps
|
|
from typing import Any, Callable, Coroutine, ParamSpec, TypeVar
|
|
import os
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# Define generic type variables for return type and parameters
|
|
R = TypeVar('R')
|
|
P = ParamSpec('P')
|
|
|
|
|
|
def time_execution_sync(additional_text: str = '') -> Callable[[Callable[P, R]], Callable[P, R]]:
|
|
def decorator(func: Callable[P, R]) -> Callable[P, R]:
|
|
@wraps(func)
|
|
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
|
|
start_time = time.time()
|
|
result = 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
|
|
|
|
|
|
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
|
|
|
|
|
|
def singleton(cls):
|
|
instance = [None]
|
|
|
|
def wrapper(*args, **kwargs):
|
|
if instance[0] is None:
|
|
instance[0] = cls(*args, **kwargs)
|
|
return instance[0]
|
|
|
|
return wrapper
|
|
|
|
|
|
def check_env_variables(keys: list[str], any_or_all=all) -> bool:
|
|
""" Check if all required environment variables are set """
|
|
return any_or_all(os.getenv(key).strip() for key in keys) |