"""Lambda util functions.""" from datetime import UTC, datetime from functools import wraps import requests from lambdacommon.common_config import logger from src.logic.retry_exception import RetriableError def log_execution(f): """Log execution details. Args: f (callable): A callable object for logging. """ @wraps(f) def _wrapper(*args, **kwargs): start_time = datetime.now(UTC) logger.info(f'function "{f.__name__}" started at {start_time} with args: {args}, kwargs: {kwargs}') try: return f(*args, **kwargs) except: # noqa logger.error(f'function "{f.__name__}" failed', exc_info=True) raise finally: execution_time = (datetime.now(UTC) - start_time).total_seconds() logger.info(f'function "{f.__name__}" took {execution_time}') return _wrapper def is_not_found_http_error(error: Exception) -> bool: """Return whether an HTTPError wraps a 404 response.""" return ( isinstance(error, requests.exceptions.HTTPError) and error.response is not None and error.response.status_code == 404 ) def detect_retriable_error(f): """Wrap logically retriable errors by RetriableError class.""" @wraps(f) def _wrapper(*args, **kwargs): try: return f(*args, **kwargs) except requests.exceptions.HTTPError as e: if is_not_found_http_error(e): raise else: raise RetriableError(e) except requests.exceptions.ConnectionError as e: raise RetriableError(e) return _wrapper