pythonintermediate

Retry Decorator with Exponential Backoff

Generic retry decorator with configurable attempts, exponential backoff, and exception filtering.

pythonPress ⌘/Ctrl + Shift + C to copy

import asyncio
import functools
import logging
from typing import Callable, Type

logger = logging.getLogger(__name__)


def retry(
    max_attempts: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    exceptions: tuple[Type[Exception], ...] = (Exception,),
) -> Callable:
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        async def async_wrapper(*args, **kwargs):
            last_exc = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return await func(*args, **kwargs)
                except exceptions as e:
                    last_exc = e
                    if attempt == max_attempts:
                        break
                    wait = delay * (backoff ** (attempt - 1))
                    logger.warning(
                        f"{func.__name__} attempt {attempt} failed: {e}. "
                        f"Retrying in {wait:.1f}s"
                    )
                    await asyncio.sleep(wait)
            raise last_exc  # type: ignore

        @functools.wraps(func)
        def sync_wrapper(*args, **kwargs):
            import time
            last_exc = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exc = e
                    if attempt == max_attempts:
                        break
                    wait = delay * (backoff ** (attempt - 1))
                    logger.warning(
                        f"{func.__name__} attempt {attempt} failed: {e}. "
                        f"Retrying in {wait:.1f}s"
                    )
                    time.sleep(wait)
            raise last_exc  # type: ignore

        if asyncio.iscoroutinefunction(func):
            return async_wrapper
        return sync_wrapper

    return decorator


# Usage:
# @retry(max_attempts=3, delay=0.5, exceptions=(ConnectionError, TimeoutError))
# async def fetch_data(url: str) -> dict:
#     async with httpx.AsyncClient() as client:
#         resp = await client.get(url)
#         return resp.json()