"""Retry decorator with exponential backoff for resilient operations.
This module provides utilities for retrying operations that may fail transiently,
with exponential backoff and jitter to prevent thundering herds.
"""
import asyncio
import functools
import logging
import time
from typing import Any, Callable, TypeVar
logger = logging.getLogger(__name__)
T = TypeVar("T")
def retry_with_backoff(
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
exceptions: tuple[type[Exception], ...] = (Exception,),
) -> Callable[[Callable[..., T]], Callable[..., T]]:
"""Decorator for retrying functions with exponential backoff.
Args:
max_retries: Maximum number of retry attempts (default: 3).
base_delay: Initial delay in seconds before first retry (default: 1.0).
max_delay: Maximum delay in seconds between retries (default: 60.0).
exceptions: Tuple of exception types to catch and retry on (default: (Exception,)).
Returns:
Decorated function that retries on specified exceptions.
Example:
@retry_with_backoff(max_retries=3, base_delay=1.0, exceptions=(IOError,))
def flaky_operation():
...
"""
def decorator(func: Callable[..., T]) -> Callable[..., T]:
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> T:
last_exception: Exception | None = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt < max_retries - 1:
delay = min(base_delay * (2**attempt), max_delay)
logger.warning(
f"{func.__name__} failed (attempt {attempt + 1}/{max_retries}): {e}. "
f"Retrying in {delay:.1f}s..."
)
time.sleep(delay)
else:
logger.error(
f"{func.__name__} failed after {max_retries} attempts: {e}"
)
if last_exception:
raise last_exception
return func(*args, **kwargs)
return wrapper
return decorator
def async_retry_with_backoff(
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
exceptions: tuple[type[Exception], ...] = (Exception,),
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
"""Decorator for retrying async functions with exponential backoff.
Args:
max_retries: Maximum number of retry attempts (default: 3).
base_delay: Initial delay in seconds before first retry (default: 1.0).
max_delay: Maximum delay in seconds between retries (default: 60.0).
exceptions: Tuple of exception types to catch and retry on (default: (Exception,)).
Returns:
Decorated async function that retries on specified exceptions.
Example:
@async_retry_with_backoff(max_retries=3, base_delay=1.0, exceptions=(IOError,))
async def flaky_operation():
...
"""
def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
@functools.wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any:
last_exception: Exception | None = None
for attempt in range(max_retries):
try:
return await func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt < max_retries - 1:
delay = min(base_delay * (2**attempt), max_delay)
logger.warning(
f"{func.__name__} failed (attempt {attempt + 1}/{max_retries}): {e}. "
f"Retrying in {delay:.1f}s..."
)
await asyncio.sleep(delay)
else:
logger.error(
f"{func.__name__} failed after {max_retries} attempts: {e}"
)
if last_exception:
raise last_exception
return await func(*args, **kwargs)
return wrapper
return decorator
