Home¶

Overview¶

aresilient is a Python library that provides resilient HTTP request functionality with automatic retry logic and exponential backoff. Built on top of the modern httpx library, it simplifies handling transient failures in HTTP communications, making your applications more robust and fault-tolerant.

Key Features¶

Automatic Retry Logic: Automatically retries failed requests for configurable HTTP status codes (429, 500, 502, 503, 504 by default)
Multiple Backoff Strategies: Choose from Exponential (default), Linear, Fibonacci, or Constant backoff, or implement your own custom strategy
Optional Jitter: Add randomized jitter to prevent thundering herd problems and avoid overwhelming servers
Retry-After Header Support: Respects server-specified retry delays from Retry-After headers (supports both integer seconds and HTTP-date formats)
Circuit Breaker Pattern: Prevent cascading failures with built-in circuit breaker support (CLOSED, OPEN, HALF_OPEN states)
Context Manager API: Manage multiple requests with shared configuration using ResilientClient and AsyncResilientClient
Complete HTTP Method Support: Supports all common HTTP methods (GET, POST, PUT, DELETE, PATCH, HEAD, OPTIONS)
Full Async Support: Fully supports asynchronous requests for high-performance applications
Built on httpx: Leverages the modern, async-capable httpx library
Advanced Configuration: Customize timeout, retry attempts, backoff strategies, jitter, retryable status codes, max total time, and max wait time
Custom Retry Predicates: Define custom logic for retry decisions based on response content, headers, or business rules using the retry_if parameter
Callbacks for Observability: Built-in callback system for logging, metrics, and alerting (on_request, on_retry, on_success, on_failure)
Type-Safe: Fully typed with comprehensive type hints
Well-Tested: Extensive test coverage ensuring reliability

Quick Examples¶

Basic GET Request¶

from aresilient import get

# Simple GET request with automatic retry
response = get("https://api.example.com/data")
print(response.json())

Basic POST Request¶

from aresilient import post

# POST request with JSON payload
response = post("https://api.example.com/submit", json={"key": "value"})
print(response.status_code)

Customizing Retry Behavior¶

from aresilient import get
from aresilient.backoff import ExponentialBackoff

# Custom retry configuration
response = get(
    "https://api.example.com/data",
    max_retries=5,  # Retry up to 5 times
    backoff_strategy=ExponentialBackoff(base_delay=1.0),  # Exponential backoff
    timeout=30.0,  # 30 second timeout
    status_forcelist=(429, 503),  # Only retry on these status codes
)

Using Different Backoff Strategies¶

from aresilient import get
from aresilient.backoff import FibonacciBackoff, LinearBackoff

# Linear backoff (1s, 2s, 3s, 4s...)
response = get(
    "https://api.example.com/data",
    backoff_strategy=LinearBackoff(base_delay=1.0, max_delay=10.0),
)

# Fibonacci backoff (1s, 1s, 2s, 3s, 5s, 8s...)
response = get(
    "https://api.example.com/data",
    backoff_strategy=FibonacciBackoff(base_delay=1.0, max_delay=10.0),
)

Using Context Manager for Multiple Requests¶

from aresilient import ResilientClient
from aresilient.core.config import ClientConfig

# Manage multiple requests with shared configuration
with ResilientClient(config=ClientConfig(max_retries=5), timeout=30.0) as client:
    response1 = client.get("https://api.example.com/users")
    response2 = client.post("https://api.example.com/data", json={"key": "value"})
    response3 = client.put(
        "https://api.example.com/resource/123", json={"status": "active"}
    )

Using Circuit Breaker¶

from aresilient import get
from aresilient.circuit_breaker import CircuitBreaker

# Prevent cascading failures with circuit breaker
circuit_breaker = CircuitBreaker(
    failure_threshold=5,  # Open circuit after 5 consecutive failures
    recovery_timeout=60.0,  # Try again after 60 seconds
)

response = get(
    "https://api.example.com/data",
    circuit_breaker=circuit_breaker,
)

Using Async¶

import asyncio
from aresilient import get_async


async def fetch_data():
    response = await get_async("https://api.example.com/data")
    return response.json()


# Run the async function
data = asyncio.run(fetch_data())
print(data)

Async with Multiple Requests¶

import asyncio
from aresilient import get_async


async def fetch_multiple():
    urls = [
        "https://api.example.com/data1",
        "https://api.example.com/data2",
        "https://api.example.com/data3",
    ]
    tasks = [get_async(url) for url in urls]
    responses = await asyncio.gather(*tasks)
    return [r.json() for r in responses]


# Fetch multiple URLs concurrently
results = asyncio.run(fetch_multiple())

Installation¶

Install using uv (recommended):

uv pip install aresilient

Or using pip:

pip install aresilient

API stability¶

While aresilient is in development stage, no API is guaranteed to be stable from one release to the next. In fact, it is very likely that the API will change multiple times before a stable 1.0.0 release. In practice, this means that upgrading aresilient to a new version will possibly break any code that was using the old version of aresilient.

License¶

aresilient is licensed under BSD 3-Clause "New" or "Revised" license available in LICENSE file.