Stamina: Production-Grade Retries
Stamina is a Python library that provides production-grade retries made easy, acting as an opinionated wrapper around Tenacity. It ensures resilient systems by handling transient failures with sensible defaults like exponential backoff with jitter, configurable exception handling, and automatic async support. It releases regularly, typically every few months, and is designed to minimize potential for misuse.
Warnings
- breaking Stamina versions 25.2.0 and newer require Python 3.10 or higher. Projects on older Python versions (3.8, 3.9) must update their Python environment before upgrading to the latest stamina versions.
- gotcha The `on` parameter in `stamina.retry()` is mandatory and has no default value. Unlike some other retry libraries, `stamina` explicitly forces you to define which exceptions trigger a retry, preventing accidental retries on non-transient errors (e.g., `Exception` generally).
- gotcha When testing retry logic, `stamina.set_testing()` is now recommended to be used as a context manager. While older global activation might still work, using it as a context manager ensures proper cleanup and isolated test environments, preventing test interference.
- gotcha Support for retrying generator functions and async generator functions (introduced in 25.2.0) is provisional and may be removed in future versions if it causes unforeseen complexities.
- gotcha Versions prior to 24.1.0 contained a bug where deactivating `stamina` globally (e.g., for testing) might not have worked correctly, leading to decorated blocks being retried even when deactivated.
Install
-
pip install stamina -
pip install stamina httpx
Imports
- retry
from stamina import retry
- retry_context
from stamina import retry_context
- set_testing
from stamina import set_testing
- RetryingCaller
from stamina import RetryingCaller
Quickstart
import httpx
import stamina
import os
@stamina.retry(on=httpx.HTTPStatusError, attempts=3, timeout=5)
def fetch_url_with_retry(url: str) -> str:
try:
response = httpx.get(url, timeout=1)
response.raise_for_status() # Raises HTTPStatusError for 4xx/5xx responses
return f"Successfully fetched: {response.status_code}"
except httpx.HTTPStatusError as e:
# Example: only retry on 5xx errors
if e.response.status_code >= 500:
print(f"Retrying on {e.response.status_code} for {url}...")
raise
else:
print(f"Not retrying on {e.response.status_code} for {url}.")
raise
except httpx.RequestError as e:
print(f"Request error for {url}: {e}, retrying...")
raise
# Example usage:
try:
# This URL will fail with 500 and retry
print(fetch_url_with_retry("https://httpbin.org/status/500"))
except (httpx.HTTPStatusError, httpx.RequestError) as e:
print(f"Final failure after retries for 500: {e}")
try:
# This URL will fail with 404 and not retry (due to the if e.response.status_code >= 500 check)
print(fetch_url_with_retry("https://httpbin.org/status/404"))
except (httpx.HTTPStatusError, httpx.RequestError) as e:
print(f"Final failure for 404 (not retried): {e}")
try:
# This URL will succeed
print(fetch_url_with_retry("https://httpbin.org/status/200"))
except (httpx.HTTPStatusError, httpx.RequestError) as e:
print(f"Unexpected failure for 200: {e}")