""" retry.py — Exponential backoff retry decorator for API calls. Usage: @retry(max_attempts=3, backoff_factor=2) async def call_api(url: str) -> dict: async with aiohttp.ClientSession() as session: async with session.get(url) as resp: resp.raise_for_status() return await resp.json() """ import asyncio import functools import logging from typing import Type logger = logging.getLogger(__name__) def retry( max_attempts: int = 3, backoff_factor: float = 2.0, initial_delay: float = 1.0, exceptions: tuple[Type[Exception], ...] = (Exception,), ): """Decorator for exponential backoff retry with jitter.""" def decorator(func): @functools.wraps(func) async def wrapper(*args, **kwargs): delay = initial_delay last_exception = None for attempt in range(1, max_attempts + 1): try: return await func(*args, **kwargs) except exceptions as e: last_exception = e if attempt == max_attempts: logger.error(f"{func.__name__} failed after {max_attempts} attempts: {e}") raise jitter = delay * 0.1 * (2 * asyncio.get_event_loop().time() % 1 - 0.5) sleep_time = delay + jitter logger.warning( f"{func.__name__} attempt {attempt}/{max_attempts} failed: {e}. " f"Retrying in {sleep_time:.1f}s..." ) await asyncio.sleep(sleep_time) delay *= backoff_factor raise last_exception # unreachable but satisfies type checker return wrapper return decorator
Production-ready async retry decorator with exponential backoff and jitter. Useful for API calls, webhook delivery, and any unreliable network operation.