# utils/retry_mechanism.py import time import logging from typing import Callable, Any, Tuple # Configure logger for this module logger = logging.getLogger(__name__) class RetryMechanism: """External retry mechanism with exponential backoff""" @staticmethod def retry_with_backoff( func: Callable, max_retries: int = 3, base_delay: float = 1.0, exceptions: Tuple[type[Exception], ...] = (Exception,) # More specific type hint ) -> Any: """ Retries a function call with exponential backoff. Args: func: The function to call. max_retries: Maximum number of retries. base_delay: Base delay in seconds for backoff. exceptions: A tuple of exception types to catch and retry on. Returns: The result of the function call if successful. Raises: The last exception encountered if all retries fail. """ last_exception = None current_delay = base_delay for attempt in range(max_retries + 1): # +1 for initial attempt try: logger.info(f"Attempt {attempt + 1}/{max_retries + 1} for function {func.__name__}") result = func() if attempt > 0: # Log if a retry was successful logger.info(f"Function {func.__name__} succeeded on attempt {attempt + 1}") return result except exceptions as e: last_exception = e logger.warning(f"Attempt {attempt + 1} for {func.__name__} failed: {str(e)}") if attempt < max_retries: logger.info(f"Waiting {current_delay:.2f} seconds before retrying {func.__name__}...") time.sleep(current_delay) current_delay *= 2 # Exponential backoff else: logger.error(f"All {max_retries + 1} attempts for {func.__name__} failed.") # If loop finishes, all retries failed, raise the last exception if last_exception is not None: raise last_exception else: # This case should ideally not be reached if func always raises on failure # or returns successfully. Added for completeness. raise RuntimeError(f"Function {func.__name__} failed after all retries without a specific exception.")