philiprehberger-retry-kit 0.4.0

Async retry with exponential backoff, circuit breaker, and cancellation for Python

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, aggressive, 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))
result = retry(lambda: transient_op(), **aggressive(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())

Resetting a Circuit Breaker

breaker = CircuitBreaker(failure_threshold=2, reset_timeout=10.0)

# ...after failures have tripped the breaker open
breaker.reset()  # Clear failure count and return to CLOSED state
result = breaker.call(lambda: 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.reset()	Reset the circuit to CLOSED state and clear failure count
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
aggressive(jitter=False)	Preset function returning aggressive config (short delays, many attempts), with optional jitter

Development

pip install -e .
python -m pytest tests/ -v

Support

If you find this project useful:

⭐ Star the repo

🐛 Report issues

💡 Suggest features

❤️ Sponsor development

🌐 All Open Source Projects

💻 GitHub Profile

🔗 LinkedIn Profile

License

MIT