Approach: implement a decorator factory retry(n, base=0.1, factor=2, jitter=0.1, exceptions=(Exception,), max_delay=None) that detects whether the wrapped function is async and runs appropriate loop. Use exponential backoff with optional random jitter and functools.wraps to preserve metadata.python
import asyncio
import functools
import inspect
import random
import time
from typing import Callable, Tuple, Any
def retry(
attempts: int = 3,
base: float = 0.1,
factor: float = 2.0,
jitter: float = 0.1,
exceptions: Tuple = (Exception,),
max_delay: float = None,
):
def decorator(fn: Callable):
is_async = inspect.iscoroutinefunction(fn)
@functools.wraps(fn)
async def _async_wrapper(*args, **kwargs):
last_exc = None
for i in range(attempts):
try:
return await fn(*args, **kwargs)
except exceptions as e:
last_exc = e
if i == attempts - 1:
break
delay = base * (factor ** i)
if jitter:
delay += random.uniform(-jitter * delay, jitter * delay)
if max_delay is not None:
delay = min(delay, max_delay)
await asyncio.sleep(max(0, delay))
raise last_exc
@functools.wraps(fn)
def _sync_wrapper(*args, **kwargs):
last_exc = None
for i in range(attempts):
try:
return fn(*args, **kwargs)
except exceptions as e:
last_exc = e
if i == attempts - 1:
break
delay = base * (factor ** i)
if jitter:
delay += random.uniform(-jitter * delay, jitter * delay)
if max_delay is not None:
delay = min(delay, max_delay)
time.sleep(max(0, delay))
raise last_exc
return _async_wrapper if is_async else _sync_wrapper
return decorator
# Example usage:
# @retry(attempts=5, base=0.2, factor=2, jitter=0.25)
# def flaky():
# ...
# @retry(attempts=4)
# async def async_flaky():
# ...
