backoff decorates Python functions with exponential backoff retry logic. pip install backoff. Exception: import backoff; @backoff.on_exception(backoff.expo, requests.exceptions.RequestException). Multiple: @backoff.on_exception(backoff.expo, (ValueError, IOError)). Max tries: @backoff.on_exception(backoff.expo, Exception, max_tries=5). Max time: @backoff.on_exception(backoff.expo, Exception, max_time=60). Predicate: @backoff.on_predicate(backoff.expo, lambda r: r.status_code == 429, max_tries=8). Expo: backoff.expo — base=2, factor=1: delays 1,2,4,8… seconds. backoff.expo(base=2, factor=0.5). Fibo: backoff.fibo — 1,1,2,3,5,8… Constant: backoff.constant(interval=2) — retry every 2 seconds. Jitter: jitter=backoff.full_jitter (default) — uniform random up to delay. jitter=backoff.random_jitter — Gaussian. jitter=None — no jitter. Callbacks: on_backoff=lambda d: print(d) on_success=... on_giveup=.... Details dict: d["tries"] d["elapsed"] d["wait"] d["exception"]. Raise: on giveup, last exception is re-raised. Async: works with async def — just add the decorator. @backoff.on_exception(backoff.expo, aiohttp.ClientError, max_tries=5). Rate limit: @backoff.on_exception(backoff.expo, RateLimitError, max_time=120, jitter=backoff.full_jitter). Claude Code generates backoff HTTP clients, rate-limit handlers, and resilient API wrappers.
CLAUDE.md for backoff
## backoff Stack
- Version: backoff >= 2.2 | pip install backoff
- Exception: @backoff.on_exception(backoff.expo, RequestException, max_tries=5)
- Predicate: @backoff.on_predicate(backoff.expo, lambda r: r is None, max_time=30)
- Wait: backoff.expo (geometric) | backoff.fibo (Fibonacci) | backoff.constant
- Limits: max_tries=N | max_time=seconds (wall-clock limit)
- Jitter: jitter=backoff.full_jitter (default) | jitter=None for exact delays
- Callbacks: on_backoff=fn | on_giveup=fn | on_success=fn
- Async: works transparently with async def functionsbackoff Exponential Retry Pipeline
# app/retry_utils.py — backoff decorators, HTTP retry wrappers, and rate-limit handlers
from __future__ import annotations
import logging
import time
from typing import Any, Callable
import backoff
logger = logging.getLogger(__name__)
# ─────────────────────────────────────────────────────────────────────────────
# 1. Callbacks
# ─────────────────────────────────────────────────────────────────────────────
def _log_backoff(details: dict) -> None:
"""Standard backoff callback: log the retry attempt."""
exc = details.get("exception")
logger.warning(
"Retrying %s (attempt %d, %.1fs elapsed, waiting %.2fs): %s",
details["target"].__name__,
details["tries"],
details["elapsed"],
details["wait"],
exc,
)
def _log_giveup(details: dict) -> None:
"""Standard giveup callback: log the final failure."""
exc = details.get("exception")
logger.error(
"Gave up %s after %d tries (%.1fs): %s",
details["target"].__name__,
details["tries"],
details["elapsed"],
exc,
)
def _log_success(details: dict) -> None:
"""Success callback: log if the function succeeded after retries."""
if details["tries"] > 1:
logger.info(
"%s succeeded on try %d (%.1fs elapsed)",
details["target"].__name__,
details["tries"],
details["elapsed"],
)
# ─────────────────────────────────────────────────────────────────────────────
# 2. Reusable decorator factories
# ─────────────────────────────────────────────────────────────────────────────
def retry_on_exception(
*exceptions,
max_tries: int = 5,
max_time: float | None = None,
base: float = 2,
factor: float = 1,
jitter=backoff.full_jitter,
log: bool = True,
):
"""
Return a decorator that retries on any of the given exceptions.
Uses exponential backoff: delay = factor * base^n (+ jitter).
max_tries and max_time are both respected — whichever triggers first.
"""
kwargs = dict(
wait_gen=backoff.expo(base=base, factor=factor),
exception=exceptions,
max_tries=max_tries,
jitter=jitter,
)
if max_time is not None:
kwargs["max_time"] = max_time
if log:
kwargs.update(on_backoff=_log_backoff, on_giveup=_log_giveup)
return backoff.on_exception(**kwargs)
def retry_on_predicate(
predicate: Callable,
max_tries: int = 5,
max_time: float | None = None,
base: float = 2,
factor: float = 1,
jitter=backoff.full_jitter,
log: bool = True,
):
"""
Return a decorator that retries when `predicate(return_value)` is truthy.
Use for polling patterns: retry while result is None or status is "pending".
"""
kwargs = dict(
wait_gen=backoff.expo(base=base, factor=factor),
predicate=predicate,
max_tries=max_tries,
jitter=jitter,
)
if max_time is not None:
kwargs["max_time"] = max_time
if log:
kwargs["on_backoff"] = _log_backoff
return backoff.on_predicate(**kwargs)
# ─────────────────────────────────────────────────────────────────────────────
# 3. Pre-built decorator presets
# ─────────────────────────────────────────────────────────────────────────────
# HTTP / network errors
def http_retry(max_tries: int = 5, max_time: float = 60):
"""Decorator: retry on common HTTP/network exceptions."""
try:
import requests
network_errors = (requests.exceptions.ConnectionError,
requests.exceptions.Timeout,
requests.exceptions.HTTPError)
except ImportError:
import urllib.error
network_errors = (urllib.error.URLError, OSError)
return retry_on_exception(*network_errors, max_tries=max_tries, max_time=max_time)
def rate_limit_retry(max_time: float = 120, max_tries: int = 10):
"""
Decorator: retry on rate-limit responses.
Works with custom RateLimitError — raise it in your request wrapper
when you see HTTP 429 or 503.
"""
return retry_on_exception(
RateLimitError,
max_tries=max_tries,
max_time=max_time,
base=2,
factor=2, # start with 2s delay: 2, 4, 8, 16...
jitter=backoff.full_jitter,
)
def transient_retry(max_tries: int = 3, interval: float = 1.0):
"""Decorator: retry with constant interval for transient errors."""
return backoff.on_exception(
backoff.constant(interval=interval),
Exception,
max_tries=max_tries,
jitter=None,
on_backoff=_log_backoff,
)
# ─────────────────────────────────────────────────────────────────────────────
# 4. Custom exception classes
# ─────────────────────────────────────────────────────────────────────────────
class RateLimitError(Exception):
"""Raised when an API returns HTTP 429 or a custom rate-limit signal."""
def __init__(self, retry_after: float | None = None):
self.retry_after = retry_after
super().__init__(f"Rate limited (retry_after={retry_after})")
class TransientError(Exception):
"""Raised for errors that are expected to resolve on retry."""
class ServerError(Exception):
"""Raised for 5xx HTTP responses."""
# ─────────────────────────────────────────────────────────────────────────────
# 5. HTTP request wrapper with retry
# ─────────────────────────────────────────────────────────────────────────────
def make_http_get(
max_tries: int = 5,
max_time: float = 60.0,
retry_statuses: tuple[int, ...] = (429, 500, 502, 503, 504),
):
"""
Return a get() function that retries on transient HTTP errors.
Raises RateLimitError on 429 so rate_limit_retry() can be used.
"""
import requests
@backoff.on_exception(
backoff.expo,
(requests.exceptions.ConnectionError,
requests.exceptions.Timeout,
RateLimitError,
ServerError),
max_tries=max_tries,
max_time=max_time,
on_backoff=_log_backoff,
on_giveup=_log_giveup,
on_success=_log_success,
)
def get(url: str, **kwargs) -> "requests.Response":
resp = requests.get(url, timeout=kwargs.pop("timeout", 10), **kwargs)
if resp.status_code == 429:
retry_after = float(resp.headers.get("Retry-After", 0))
raise RateLimitError(retry_after=retry_after)
if resp.status_code in (500, 502, 503, 504):
raise ServerError(f"HTTP {resp.status_code}: {url}")
resp.raise_for_status()
return resp
return get
# ─────────────────────────────────────────────────────────────────────────────
# 6. Async variants
# ─────────────────────────────────────────────────────────────────────────────
async def async_retry_example():
"""backoff works transparently with async functions."""
import asyncio
@backoff.on_exception(
backoff.expo,
(IOError, RuntimeError),
max_tries=4,
on_backoff=_log_backoff,
)
async def flaky_fetch(url: str) -> str:
# Simulated async operation
await asyncio.sleep(0.01)
return f"data from {url}"
return await flaky_fetch("https://example.com")
# ─────────────────────────────────────────────────────────────────────────────
# 7. Polling with on_predicate
# ─────────────────────────────────────────────────────────────────────────────
def poll_until_ready(
check_fn: Callable[[], Any],
predicate: Callable[[Any], bool] | None = None,
max_tries: int = 20,
max_time: float = 120.0,
) -> Any:
"""
Poll `check_fn()` until `predicate(result)` is falsy (i.e., ready).
Default predicate: retry while result is None.
"""
if predicate is None:
predicate = lambda r: r is None
@backoff.on_predicate(
backoff.expo,
predicate=predicate,
max_tries=max_tries,
max_time=max_time,
jitter=backoff.full_jitter,
on_backoff=_log_backoff,
)
def _poll():
return check_fn()
return _poll()
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
import random
call_count = [0]
# ── Demo 1: on_exception with attempt counter ──────────────────────────
@backoff.on_exception(
backoff.expo,
(ValueError,),
max_tries=4,
jitter=None,
on_backoff=lambda d: print(f" Retry #{d['tries']} (wait {d['wait']:.2f}s)"),
)
def flaky_function():
call_count[0] += 1
if call_count[0] < 3:
raise ValueError(f"Transient error #{call_count[0]}")
return f"success on call #{call_count[0]}"
print("=== on_exception (succeeds on 3rd try) ===")
result = flaky_function()
print(f" Result: {result}")
# ── Demo 2: on_predicate ───────────────────────────────────────────────
poll_count = [0]
@backoff.on_predicate(
backoff.expo,
lambda r: r != "done",
max_tries=5,
jitter=None,
on_backoff=lambda d: print(f" Poll #{d['tries']}: not ready yet"),
)
def check_status():
poll_count[0] += 1
return "done" if poll_count[0] >= 3 else "pending"
print("\n=== on_predicate (polling) ===")
status = check_status()
print(f" Final status: {status}")
# ── Demo 3: giveup after max_tries ─────────────────────────────────────
print("\n=== on_exception (giveup) ===")
giveup_called = [False]
@backoff.on_exception(
backoff.constant(interval=0.01),
RuntimeError,
max_tries=3,
jitter=None,
on_giveup=lambda d: giveup_called.__setitem__(0, True),
)
def always_fails():
raise RuntimeError("always bad")
try:
always_fails()
except RuntimeError:
print(f" RuntimeError raised after 3 tries | on_giveup called: {giveup_called[0]}")
# ── Demo 4: Fibonacci backoff ──────────────────────────────────────────
fib_waits = []
@backoff.on_exception(
backoff.fibo,
ValueError,
max_tries=6,
jitter=None,
on_backoff=lambda d: fib_waits.append(round(d["wait"], 2)),
)
def fibonacci_target():
raise ValueError("trigger fib")
try:
fibonacci_target()
except ValueError:
pass
print(f"\n=== Fibonacci wait sequence ===\n {fib_waits}")For the tenacity alternative — tenacity and backoff both provide retry logic via decorators; backoff uses a functional/generator-based API (backoff.expo, backoff.fibo) and is more concise for simple cases; tenacity uses a richer DSL with retry_if_exception_type, wait_exponential, stop_after_attempt, before_sleep all composable via | operators — use tenacity when you need combine multiple stop/wait/retry strategies, and backoff when the decorator-first API is more readable. For the retry (retry package) alternative — the retry package (pip install retry) provides @retry and @retry(tries=3, delay=2, backoff=2) but is less maintained and lacks async support, predicate-based retry, and separate jitter configurations; backoff is more actively maintained and covers async, full_jitter, giveup callbacks, and Fibonacci/constant wait generators. The Claude Skills 360 bundle includes backoff skill sets covering @backoff.on_exception with exception tuple, @backoff.on_predicate for polling, backoff.expo/fibo/constant wait generators, max_tries/max_time limits, full_jitter/random_jitter, on_backoff/on_giveup/on_success callbacks, retry_on_exception() and retry_on_predicate() decorator factories, http_retry() and rate_limit_retry() presets, RateLimitError/TransientError/ServerError custom exceptions, make_http_get() HTTP retry wrapper, poll_until_ready() polling helper, and async function support. Start with the free tier to try exponential backoff retry code generation.