Retry Decorator With Exponential Backoff And Jitter
A typed retry decorator that retries only transient failures, caps exponential backoff and adds jitter to avoid thundering herds.
Code
import random
import time
from functools import wraps
from typing import Callable, Tuple, Type, TypeVar
T = TypeVar("T")
class RetryConfigError(ValueError):
"""Raised when retry parameters are invalid."""
def retry(
max_attempts: int = 3,
base_delay_s: float = 0.1,
cap_s: float = 5.0,
jitter_ratio: float = 0.1,
retryable: Tuple[Type[BaseException], ...] = (Exception,),
sleep: Callable[[float], None] = time.sleep,
) -> Callable[[Callable[..., T]], Callable[..., T]]:
"""Return a decorator that retries transient failures with jittered backoff."""
if max_attempts < 1:
raise RetryConfigError("max_attempts must be >= 1")
if base_delay_s <= 0 or cap_s < base_delay_s:
raise RetryConfigError("base_delay_s must be > 0 and <= cap_s")
if not 0 <= jitter_ratio <= 1:
raise RetryConfigError("jitter_ratio must be within [0, 1]")
def decorator(fn: Callable[..., T]) -> Callable[..., T]:
@wraps(fn)
def wrapper(*args: object, **kwargs: object) -> T:
last_error: BaseException | None = None
for attempt in range(max_attempts):
try:
return fn(*args, **kwargs)
except retryable as exc:
last_error = exc
if attempt == max_attempts - 1:
break
delay = min(cap_s, base_delay_s * (2 ** attempt))
jitter = random.uniform(0, delay * jitter_ratio)
sleep(delay + jitter)
assert last_error is not None
raise last_error
return wrapper
return decoratorReport a problem with this exampleOpens a pre-filled issue on GitHub.