Async Retry Decorator

The problem

You need to retry async operations that might fail temporarily (async API calls, async database queries).

The solution

Python

import asyncio
import functools
from typing import Tuple, Type

def async_retry(
    max_attempts: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    exceptions: Tuple[Type[Exception], ...] = (Exception,)
):
    """Async retry decorator with exponential backoff."""
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            current_delay = delay
            last_exception = None
            
            for attempt in range(max_attempts):
                try:
                    return await func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        await asyncio.sleep(current_delay)
                        current_delay *= backoff
            
            raise last_exception
        return wrapper
    return decorator

Parameters

max_attemptsint

Maximum number of retry attempts

delayfloat

Initial delay between retries in seconds

backofffloat

Multiplier for delay after each retry

exceptionstuple

Exception types to catch and retry

Put it to work

Example

import aiohttp

@async_retry(max_attempts=3, delay=1, exceptions=(aiohttp.ClientError,))
async def fetch_async(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

# Usage in async context
data = await fetch_async('https://api.example.com/data')

Retry Decorator with Backoff

Retry Decorator with Jitter

TTL Cache Decorator