Spaces:
Running
Running
| # 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""" | |
| 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.") | |