philiprehberger-retry-kit
v0.3.1204 downloadsMIT
Async retry with exponential backoff, circuit breaker, and cancellation for Python
$
pip install philiprehberger-retry-kitREADME
philiprehberger-retry-kit
Retry with exponential backoff, circuit breaker, and presets for Python.
Installation
pip install philiprehberger-retry-kit
Usage
Basic Retry
from philiprehberger_retry_kit import retry
result = retry(lambda: fetch_data(), max_attempts=3)
With Options
result = retry(
lambda: fetch_data(),
max_attempts=5,
backoff="exponential", # "exponential" | "linear" | "fixed"
initial_delay=1.0,
max_delay=30.0,
jitter=True,
retry_on=lambda e: isinstance(e, ConnectionError),
on_retry=lambda e, attempt: print(f"Retry {attempt}..."),
)
Async Retry
from philiprehberger_retry_kit import async_retry
result = await async_retry(
lambda: async_fetch_data(),
max_attempts=3,
backoff="exponential",
)
Presets
from philiprehberger_retry_kit import retry, presets
result = retry(lambda: fetch_data(), **presets["network_request"])
result = retry(lambda: db_query(), **presets["database_query"])
result = retry(lambda: critical_op(), **presets["aggressive"])
result = retry(lambda: gentle_op(), **presets["gentle"])
Preset Functions with Jitter
Preset functions accept a jitter parameter that randomizes the initial delay by a factor between 0.8 and 1.2, helping to spread out retry storms.
from philiprehberger_retry_kit import exponential, gentle, network_request, database_query, retry
result = retry(lambda: fetch_data(), **exponential(jitter=True))
result = retry(lambda: gentle_op(), **gentle(jitter=True))
result = retry(lambda: api_call(), **network_request(jitter=True))
result = retry(lambda: db_query(), **database_query(jitter=True))
Circuit Breaker
from philiprehberger_retry_kit import CircuitBreaker, CircuitOpenError
breaker = CircuitBreaker(
failure_threshold=5,
reset_timeout=30.0,
on_state_change=lambda from_s, to_s: print(f"Circuit: {from_s} -> {to_s}"),
)
try:
result = breaker.call(lambda: fetch_data())
except CircuitOpenError:
print("Circuit is open, failing fast")
Circuit Breaker as Context Manager
breaker = CircuitBreaker(failure_threshold=3, reset_timeout=10.0)
try:
with breaker:
result = fetch_data()
except CircuitOpenError:
print("Circuit is open")
On __enter__, the breaker checks if the circuit is open and raises CircuitOpenError if so. On __exit__, it records success or failure based on whether an exception occurred.
Async Circuit Breaker
result = await breaker.async_call(lambda: async_fetch_data())
API
| Function / Class | Description |
|---|---|
retry(fn, *, max_attempts=3, backoff="exponential", initial_delay=1.0, max_delay=30.0, jitter=True, retry_on=None, on_retry=None, on_success=None, on_failure=None) | Retry a callable with configurable backoff |
async_retry(fn, *, ...) | Async version of retry() with the same parameters |
CircuitBreaker(failure_threshold=5, reset_timeout=30.0, half_open_max_attempts=1, on_state_change=None, on_circuit_open=None) | Circuit breaker that fails fast after repeated failures |
CircuitBreaker.call(fn) | Execute function through the circuit breaker |
CircuitBreaker.async_call(fn) | Async version of call() |
CircuitBreaker.__enter__ / __exit__ | Context manager support for circuit breaker |
RetryError | Raised when all retry attempts fail (.attempts, .last_error) |
CircuitOpenError | Raised when circuit breaker is open |
presets | Dict of preset configs: "aggressive", "gentle", "network_request", "database_query" |
exponential(jitter=False) | Preset function returning aggressive config, with optional jitter |
gentle(jitter=False) | Preset function returning gentle config, with optional jitter |
network_request(jitter=False) | Preset function returning network request config, with optional jitter |
database_query(jitter=False) | Preset function returning database query config, with optional jitter |
Development
pip install -e .
python -m pytest tests/ -v
Support
If you find this project useful: