58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
"""
|
|
Retry utility with exponential backoff for failed requests.
|
|
"""
|
|
import time
|
|
import functools
|
|
from typing import Callable, Type, Tuple
|
|
from utils.logger import setup_logger
|
|
|
|
logger = setup_logger(__name__)
|
|
|
|
|
|
def retry_with_backoff(
|
|
max_retries: int = 3,
|
|
base_delay: float = 1.0,
|
|
max_delay: float = 60.0,
|
|
exponential_base: float = 2.0,
|
|
exceptions: Tuple[Type[Exception], ...] = (Exception,)
|
|
):
|
|
"""
|
|
Decorator to retry a function with exponential backoff.
|
|
|
|
Args:
|
|
max_retries: Maximum number of retry attempts
|
|
base_delay: Initial delay between retries in seconds
|
|
max_delay: Maximum delay between retries
|
|
exponential_base: Base for exponential backoff calculation
|
|
exceptions: Tuple of exception types to catch and retry
|
|
|
|
Returns:
|
|
Decorated function with retry logic
|
|
"""
|
|
def decorator(func: Callable):
|
|
@functools.wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
retries = 0
|
|
while retries <= max_retries:
|
|
try:
|
|
return func(*args, **kwargs)
|
|
except exceptions as e:
|
|
retries += 1
|
|
if retries > max_retries:
|
|
logger.error(
|
|
f"Function {func.__name__} failed after {max_retries} retries. "
|
|
f"Error: {str(e)}"
|
|
)
|
|
raise
|
|
|
|
delay = min(base_delay * (exponential_base ** (retries - 1)), max_delay)
|
|
logger.warning(
|
|
f"Function {func.__name__} failed (attempt {retries}/{max_retries}). "
|
|
f"Retrying in {delay:.2f} seconds. Error: {str(e)}"
|
|
)
|
|
time.sleep(delay)
|
|
|
|
return None
|
|
return wrapper
|
|
return decorator
|
|
|