Python Senior (5–8 yrs) Coding Frequent Python asyncio retry exponential backoff async

Implement an async retry function with exponential backoff in Python (asyncio)

py-sen-002

Your answer

Answer as you would in a real interview — explain your thinking, not just the conclusion.

Model answer

The async version uses asyncio.sleep (non-blocking) instead of time.sleep. Wrap in a decorator using @functools.wraps and inspect whether the wrapped function is a coroutine function with asyncio.iscoroutinefunction. A more reusable approach: accept an async function directly. Add a timeout per attempt using asyncio.wait_for to avoid hanging forever on a single attempt. Add jitter to prevent thundering herd. Return the result on success, raise on final failure.

Code example

import asyncio
import functools
import random
from collections.abc import Callable, Awaitable
from typing import TypeVar, Any

T = TypeVar("T")

def async_retry(
    max_attempts: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
    exceptions: tuple[type[Exception], ...] = (Exception,),
    per_attempt_timeout: float | None = None,
):
    def decorator(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
        @functools.wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> T:
            for attempt in range(max_attempts):
                try:
                    coro = func(*args, **kwargs)
                    if per_attempt_timeout:
                        return await asyncio.wait_for(coro, timeout=per_attempt_timeout)
                    return await coro
                except asyncio.TimeoutError as exc:
                    if attempt == max_attempts - 1:
                        raise
                    delay = min(base_delay * (2**attempt), max_delay)
                    delay += random.uniform(0, delay * 0.1)  # 10% jitter
                    await asyncio.sleep(delay)
                except exceptions as exc:
                    if attempt == max_attempts - 1:
                        raise
                    delay = min(base_delay * (2**attempt), max_delay)
                    delay += random.uniform(0, delay * 0.1)
                    await asyncio.sleep(delay)
        return wrapper
    return decorator

# Usage
import aiohttp

@async_retry(max_attempts=4, base_delay=0.5, per_attempt_timeout=5.0,
             exceptions=(aiohttp.ClientError,))
async def fetch_json(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            resp.raise_for_status()
            return await resp.json()

Follow-up

How would you implement circuit-breaker logic on top of this retry mechanism — opening the circuit after N consecutive failures and half-opening after a cooldown?