Tenacity provides powerful retry logic for Python functions. pip install tenacity. from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type. Basic: @retry(stop=stop_after_attempt(3)). Exponential backoff: @retry(wait=wait_exponential(multiplier=1, min=1, max=60)). Fixed wait: @retry(wait=wait_fixed(2)). Jitter: @retry(wait=wait_exponential_jitter(initial=1, max=60)). Combined: @retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=2, min=2, max=30)). Condition: @retry(retry=retry_if_exception_type(ConnectionError)). Multiple: @retry(retry=retry_if_exception_type((TimeoutError, ConnectionError))). Result: @retry(retry=retry_if_result(lambda r: r is None)). Not result: @retry(retry=retry_if_not_result(lambda r: r.ok)). Before sleep: from tenacity import before_sleep_log; @retry(before_sleep=before_sleep_log(logger, logging.WARNING)). Stop delay: stop=stop_after_delay(60). Combined stop: stop=stop_after_attempt(5) | stop_after_delay(120). Reraise: @retry(reraise=True) — raises original exception instead of RetryError. RetryError: from tenacity import RetryError — catch when all attempts exhausted. TryAgain: raise tenacity.TryAgain inside function to force retry. Async: from tenacity import AsyncRetrying, async with AsyncRetrying(stop=stop_after_attempt(3)) as r: async for attempt in r: with attempt: result = await coro(). Custom: @retry(after=lambda r: print(f"retry {r.attempt_number}")). Statistics: fn.retry.statistics. Claude Code generates Tenacity retry decorators for HTTP clients, database connections, and external API integrations.
CLAUDE.md for Tenacity
## Tenacity Stack
- Version: tenacity >= 8.2
- Decorator: @retry(stop=..., wait=..., retry=..., before_sleep=...)
- Stop: stop_after_attempt(N) | stop_after_delay(secs) | combine with |
- Wait: wait_fixed | wait_exponential(mult, min, max) | wait_exponential_jitter
- Retry on: retry_if_exception_type(ExcClass) | retry_if_result(fn)
- Logging: before_sleep=before_sleep_log(logger, logging.WARNING)
- Reraise: reraise=True to surface original exception (not RetryError)
- Async: AsyncRetrying context manager for coroutinesTenacity Retry Pipeline
# utils/tenacity_pipeline.py — resilient retry logic with Tenacity
from __future__ import annotations
import asyncio
import logging
import random
import time
from functools import wraps
from typing import Any, Callable, Type
import tenacity
from tenacity import (
retry,
retry_if_exception_type,
retry_if_not_result,
retry_if_result,
stop_after_attempt,
stop_after_delay,
wait_exponential,
wait_exponential_jitter,
wait_fixed,
wait_random,
before_sleep_log,
after_log,
RetryError,
TryAgain,
AsyncRetrying,
)
logger = logging.getLogger(__name__)
# ── 0. Pre-configured decorators ─────────────────────────────────────────────
def http_retry(
max_attempts: int = 5,
min_wait: float = 1.0,
max_wait: float = 60.0,
multiplier: float = 2.0,
reraise: bool = True,
):
"""
Standard HTTP retry: exponential back-off with jitter, 5 attempts max.
Retries on ConnectionError, TimeoutError, and OSError (transient network).
"""
return retry(
stop=stop_after_attempt(max_attempts),
wait=wait_exponential_jitter(initial=min_wait, max=max_wait, exp_base=multiplier),
retry=retry_if_exception_type((ConnectionError, TimeoutError, OSError)),
before_sleep=before_sleep_log(logger, logging.WARNING),
reraise=reraise,
)
def database_retry(
max_attempts: int = 3,
wait_seconds: float = 2.0,
reraise: bool = True,
):
"""
Short fixed-wait retry for transient database errors.
Retries on common operational exceptions; does NOT retry programming errors.
"""
return retry(
stop=stop_after_attempt(max_attempts),
wait=wait_fixed(wait_seconds),
retry=retry_if_exception_type((OSError, ConnectionError, TimeoutError)),
before_sleep=before_sleep_log(logger, logging.WARNING),
reraise=reraise,
)
def api_retry(
max_attempts: int = 4,
retry_on_status: tuple[int, ...] = (429, 500, 502, 503, 504),
reraise: bool = True,
):
"""
Retry decorator for HTTP API calls that return response objects.
Both raises exceptions AND retries on specific HTTP status codes.
Pass your response object — retry_if_result checks .status_code.
"""
def _should_retry_response(r) -> bool:
if r is None:
return True
status = getattr(r, "status_code", None) or getattr(r, "status", None)
return status in retry_on_status
return retry(
stop=stop_after_attempt(max_attempts),
wait=wait_exponential_jitter(initial=1, max=30),
retry=(
retry_if_exception_type((ConnectionError, TimeoutError)) |
retry_if_result(_should_retry_response)
),
before_sleep=before_sleep_log(logger, logging.WARNING),
reraise=reraise,
)
# ── 1. Async retry helpers ────────────────────────────────────────────────────
async def retry_async(
coro_fn: Callable,
*args,
max_attempts: int = 5,
min_wait: float = 1.0,
max_wait: float = 60.0,
exceptions: tuple = (Exception,),
**kwargs,
) -> Any:
"""
Retry an async coroutine with exponential back-off.
Alternative to @retry for when you can't decorate the function directly.
"""
async for attempt in AsyncRetrying(
stop=stop_after_attempt(max_attempts),
wait=wait_exponential_jitter(initial=min_wait, max=max_wait),
retry=retry_if_exception_type(exceptions),
before_sleep=before_sleep_log(logger, logging.WARNING),
reraise=True,
):
with attempt:
return await coro_fn(*args, **kwargs)
def async_http_retry(
max_attempts: int = 5,
min_wait: float = 1.0,
max_wait: float = 60.0,
):
"""Async version of http_retry decorator."""
return retry(
stop=stop_after_attempt(max_attempts),
wait=wait_exponential_jitter(initial=min_wait, max=max_wait),
retry=retry_if_exception_type((ConnectionError, TimeoutError, OSError)),
before_sleep=before_sleep_log(logger, logging.WARNING),
reraise=True,
)
# ── 2. Conditional retry patterns ─────────────────────────────────────────────
def retry_until_truthy(
fn: Callable,
*args,
max_attempts: int = 10,
wait_seconds: float = 5.0,
**kwargs,
) -> Any:
"""
Keep calling fn until it returns a truthy value.
Useful for polling: wait until job is complete, resource is available, etc.
"""
@retry(
stop=stop_after_attempt(max_attempts),
wait=wait_fixed(wait_seconds),
retry=retry_if_result(lambda r: not r),
before_sleep=before_sleep_log(logger, logging.INFO),
reraise=False,
)
def _call():
return fn(*args, **kwargs)
try:
return _call()
except RetryError:
return None
def retry_until_not_none(
fn: Callable,
*args,
max_attempts: int = 5,
wait_seconds: float = 2.0,
**kwargs,
) -> Any:
"""Poll fn until it returns a non-None value or max_attempts is exhausted."""
@retry(
stop=stop_after_attempt(max_attempts),
wait=wait_fixed(wait_seconds),
retry=retry_if_result(lambda r: r is None),
reraise=False,
)
def _call():
return fn(*args, **kwargs)
try:
return _call()
except RetryError:
return None
# ── 3. Usage examples ─────────────────────────────────────────────────────────
@http_retry(max_attempts=4, min_wait=1.5, max_wait=30.0)
def fetch_url(url: str, timeout: float = 10.0) -> dict:
"""
HTTP GET with retry. In production, replace urllib.request with httpx/requests.
"""
import urllib.request, json
with urllib.request.urlopen(url, timeout=timeout) as resp:
return json.loads(resp.read())
@database_retry(max_attempts=3, wait_seconds=1.0)
def query_database(connection, sql: str, params: tuple = ()) -> list:
"""Database query with retry on transient connection errors."""
cursor = connection.cursor()
cursor.execute(sql, params)
return cursor.fetchall()
@retry(
stop=stop_after_attempt(3) | stop_after_delay(120),
wait=wait_exponential(multiplier=1, min=2, max=30),
retry=retry_if_exception_type(Exception),
before_sleep=before_sleep_log(logger, logging.WARNING),
after=after_log(logger, logging.DEBUG),
reraise=True,
)
def call_external_service(payload: dict) -> dict:
"""
Call an external service with combined stop condition:
give up after 3 attempts OR after 2 minutes, whichever comes first.
"""
# simulate a call that sometimes fails
if random.random() < 0.4:
raise ConnectionError("Service temporarily unavailable")
return {"status": "ok", "payload": payload}
@async_http_retry(max_attempts=3, min_wait=0.5, max_wait=10.0)
async def async_fetch(session, url: str) -> dict:
"""Async HTTP fetch with retry (use with aiohttp ClientSession)."""
async with session.get(url) as response:
response.raise_for_status()
return await response.json()
# ── 4. Retry with TryAgain for custom logic ───────────────────────────────────
@retry(stop=stop_after_attempt(5), wait=wait_exponential(min=1, max=30), reraise=True)
def smart_api_call(client, resource_id: str) -> dict:
"""
Use TryAgain for manual retry signaling inside the function body.
Useful when retry condition requires parsing the response (not just exceptions).
"""
response = client.get(f"/resources/{resource_id}")
if response.status_code == 202:
# Resource still processing — retry
logger.info("Resource %s still processing, retrying...", resource_id)
raise TryAgain
if response.status_code == 200:
return response.json()
raise RuntimeError(f"Unexpected status: {response.status_code}")
# ── 5. Retry statistics ────────────────────────────────────────────────────────
def with_retry_stats(fn: Callable) -> Callable:
"""
Wrap a retried function to log retry statistics after each call.
"""
@wraps(fn)
def wrapper(*args, **kwargs):
try:
result = fn(*args, **kwargs)
stats = fn.retry.statistics
if stats.get("attempt_number", 1) > 1:
logger.info(
"%s succeeded after %d attempts (%.2fs elapsed)",
fn.__name__,
stats.get("attempt_number", 1),
stats.get("idle_for", 0),
)
return result
except RetryError as exc:
stats = fn.retry.statistics
logger.error(
"%s exhausted %d attempts: %s",
fn.__name__,
stats.get("attempt_number", "?"),
exc,
)
raise
return wrapper
# ── Demo ──────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO,
format="%(asctime)s %(levelname)-8s %(message)s")
print("Tenacity Retry Demo")
print("=" * 50)
# Simulate a flaky function
call_count = [0]
@retry(
stop=stop_after_attempt(5),
wait=wait_exponential_jitter(initial=0.1, max=1.0),
retry=retry_if_exception_type(ValueError),
before_sleep=before_sleep_log(logger, logging.INFO),
reraise=True,
)
def flaky_fn():
call_count[0] += 1
if call_count[0] < 3:
raise ValueError(f"Attempt {call_count[0]} failed")
return f"Success on attempt {call_count[0]}"
try:
result = flaky_fn()
print(f"\nResult: {result}")
print(f"Statistics: {flaky_fn.retry.statistics}")
except RetryError as e:
print(f"All attempts exhausted: {e}")
# Polling example
print("\nPolling until truthy:")
state = {"ready": False, "tick": 0}
def check_ready():
state["tick"] += 1
if state["tick"] >= 3:
state["ready"] = True
return state["ready"]
result = retry_until_truthy(check_ready, max_attempts=5, wait_seconds=0.01)
print(f" Ready after {state['tick']} poll(s): {result}")For the custom time.sleep + try/except loop alternative — manual retry loops duplicate boilerplate across every function, increment attempt counters by hand, and silently swallow exceptions when tired while Tenacity’s @retry(stop=stop_after_attempt(5), wait=wait_exponential_jitter(...)) centralizes retry policy, computes jittered back-off, logs each attempt via before_sleep_log, and surfaces the original exception with reraise=True — removing 20 lines of error-prone manual retry code with a single decorator. For the backoff library alternative — backoff has @backoff.on_exception but no retry_if_result predicate (cannot retry on HTTP 429 without raising), no combined stop conditions (stop_after_attempt | stop_after_delay), and no AsyncRetrying context manager, while Tenacity’s retry_if_result(lambda r: r.status_code in (429, 503)) retries on response status without raising, enabling the same decorator for exception-based and result-based policies. The Claude Skills 360 bundle includes Tenacity skill sets covering http_retry/database_retry/api_retry pre-built decorators, stop_after_attempt and stop_after_delay combination, wait_exponential_jitter, retry_if_exception_type / retry_if_result, before_sleep_log, TryAgain for manual retry, AsyncRetrying for coroutines, retry statistics, and retry_until_truthy polling pattern. Start with the free tier to try retry logic code generation.