philiprehberger-http-retry 0.3.0

Resilient HTTP client with automatic retries and backoff.

philiprehberger-http-retry

Resilient HTTP client with automatic retries and configurable backoff.

Installation

pip install philiprehberger-http-retry

Usage

from philiprehberger_http_retry import resilient_get, resilient_post, Session

Simple Requests

# GET request with default retries (3 attempts, exponential backoff)
response = resilient_get("https://api.example.com/data")
print(response.read().decode())

# POST with JSON body
response = resilient_post(
    "https://api.example.com/items",
    json_data={"name": "widget", "count": 5},
)

Backoff Strategies

from philiprehberger_http_retry import resilient_request

# Exponential backoff (default): 0.5s, 1s, 2s, ...
resilient_request("GET", url, backoff="exponential")

# Linear backoff: 0.5s, 1s, 1.5s, ...
resilient_request("GET", url, backoff="linear")

# Constant backoff: 0.5s, 0.5s, 0.5s, ...
resilient_request("GET", url, backoff="constant")

# Custom callable: receives attempt number, returns delay in seconds
resilient_request("GET", url, backoff=lambda attempt: 0.1 * (attempt + 1))

Retry Hook

import logging

def log_retry(attempt: int, error: Exception) -> None:
    logging.warning(f"Retry {attempt}: {error}")

response = resilient_get(
    "https://api.example.com/data",
    on_retry=log_retry,
)

Circuit breaker

from philiprehberger_http_retry import (
    CircuitBreaker,
    CircuitBreakerOpen,
    resilient_get,
)

breaker = CircuitBreaker(failure_threshold=5, reset_timeout=30.0)

try:
    response = resilient_get(
        "https://flaky-api.example.com/data",
        circuit_breaker=breaker,
    )
except CircuitBreakerOpen as err:
    print(f"Breaker tripped, retry after unix time {err.next_retry_at:.0f}")

The breaker counts request outcomes (not retry attempts). After failure_threshold consecutive failures it trips to open and rejects calls until reset_timeout elapses, then probes one request in half_open before returning to closed.

Session with Defaults

session = Session(
    base_url="https://api.example.com",
    default_headers={"Authorization": "Bearer token123"},
    retries=5,
    backoff="linear",
    timeout=15,
    on_retry=log_retry,
)

response = session.get("/users")
response = session.post("/users", json_data={"name": "Alice"})

Error Handling

from philiprehberger_http_retry import resilient_get, RetryExhaustedError

try:
    response = resilient_get("https://unreliable-api.example.com/data")
except RetryExhaustedError as err:
    print(f"Failed after {err.attempts} attempts: {err.last_error}")

API

Function / Class	Description
resilient_request(method, url, **kwargs)	Core retry function. Supports data, headers, retries, backoff, timeout, retry_on, on_retry.
resilient_get(url, **kwargs)	GET convenience wrapper around resilient_request.
resilient_post(url, data=None, json_data=None, **kwargs)	POST wrapper. Auto-serializes json_data and sets Content-Type.
Session(base_url, default_headers, retries, backoff, timeout, retry_on, on_retry, circuit_breaker)	Stores defaults. Methods: get(path), post(path).
CircuitBreaker(failure_threshold, reset_timeout, half_open_max_calls)	Trips to open after consecutive failures, probes via half_open. Methods: allow_request(), record_success(), record_failure().
CircuitBreakerOpen	Raised when the breaker rejects a request. Attribute: .next_retry_at (unix timestamp).
RetryExhaustedError	Raised after all retries fail. Attributes: .attempts, .last_error.

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