### Basic Pyresilience Installation with Uv

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/getting-started/installation.md

Install Pyresilience using the uv package manager.

```bash
uv pip install pyresilience
```

--------------------------------

### Verify Pyresilience Installation

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/getting-started/installation.md

After installation, run this Python code to import the pyresilience library and print its installed version to confirm the installation was successful.

```python
import pyresilience
print(pyresilience.__version__)  # 0.2.0
```

--------------------------------

### Install Resilience Libraries

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/getting-started/comparison.md

Installs pyresilience, tenacity, backoff, stamina, and pybreaker for performance benchmarking. Ensure you have Python 3.10+ installed.

```bash
pip install pyresilience tenacity backoff stamina pybreaker
```

--------------------------------

### Basic Pyresilience Installation with Pip

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/getting-started/installation.md

Use this command for a standard installation of Pyresilience using pip.

```bash
pip install pyresilience
```

--------------------------------

### Quick Start: Basic Resilience Decorator

Source: https://github.com/ahsansheraz/pyresilience/blob/main/README.md

Apply the resilient decorator with basic configurations for retry, timeout, and circuit breaker to a function. This example demonstrates how to protect an API call.

```python
import requests
from pyresilience import resilient, RetryConfig, TimeoutConfig, CircuitBreakerConfig

@resilient(
    retry=RetryConfig(max_attempts=3, delay=1.0),
    timeout=TimeoutConfig(seconds=10),
    circuit_breaker=CircuitBreakerConfig(failure_threshold=5),
)
def call_api(endpoint: str) -> dict:
    return requests.get(endpoint).json()
```

--------------------------------

### FastAPI Integration Setup

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Demonstrates how to integrate Pyresilience with FastAPI using ResilientDependency and ResilientMiddleware for per-route resilience.

```python
from fastapi import Depends, FastAPI
from pyresilience import (
    ResilienceConfig,
    RetryConfig,
    TimeoutConfig,
    CircuitBreakerConfig,
    RateLimiterConfig,
)
from pyresilience.contrib.fastapi import ResilientDependency, ResilientMiddleware

app = FastAPI()
```

--------------------------------

### Clone Repository and Set Up Development Environment

Source: https://github.com/ahsansheraz/pyresilience/blob/main/CONTRIBUTING.md

Clone the pyresilience repository and install development dependencies using the make dev command.

```bash
git clone https://github.com/AhsanSheraz/pyresilience.git
cd pyresilience
make dev
```

--------------------------------

### Install pyresilience

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Install the pyresilience library using pip. For optional performance backends like uvloop and orjson, use the `[fast]` extra.

```bash
pip install pyresilience
```

```bash
pip install pyresilience[fast]
```

--------------------------------

### Maximum Resilience Configuration

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/combining.md

This example demonstrates combining all available resilience patterns for maximum fault tolerance in mission-critical calls.

```python
from pyresilience import (    resilient,    CacheConfig,    RetryConfig,    TimeoutConfig,    CircuitBreakerConfig,    RateLimiterConfig,    BulkheadConfig,    FallbackConfig,) 
import requests
from pyresilience.listeners import JsonEventLogger, MetricsCollector

@resilient(
    cache=CacheConfig(max_size=256, ttl=300.0),
    retry=RetryConfig(max_attempts=3, delay=0.5, backoff_factor=2.0, jitter=True),
    timeout=TimeoutConfig(seconds=10),
    circuit_breaker=CircuitBreakerConfig(failure_threshold=5, recovery_timeout=30),
    rate_limiter=RateLimiterConfig(max_calls=100, period=60.0),
    bulkhead=BulkheadConfig(max_concurrent=20),
    fallback=FallbackConfig(handler=lambda e: None),
    listeners=[JsonEventLogger(), MetricsCollector()],
)
def mission_critical_call(request_id: str) -> dict:
    return requests.post("https://api.example.com/process", json={"id": request_id}).json()
```

--------------------------------

### Check and Install Uvloop

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/getting-started/installation.md

Use these Python functions to check if uvloop and orjson are available and to install uvloop as the default event loop policy. These functions are useful after installing with the 'fast' extras.

```python
from pyresilience import has_uvloop, has_orjson, install_uvloop

print(has_uvloop())   # True if uvloop is available
print(has_orjson())   # True if orjson is available

install_uvloop()      # Installs uvloop as the default event loop policy
```

--------------------------------

### Install Pyresilience with Performance Backends

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/getting-started/installation.md

For production environments, install Pyresilience with the 'fast' extras to include optional performance backends like uvloop and orjson. The library auto-detects these backends at runtime.

```bash
pip install pyresilience[fast]
```

--------------------------------

### Clone Repository and Install Dependencies

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/contributing.md

Clone the PyResilience repository and install development dependencies. Ensure you are in a Python virtual environment.

```bash
git clone https://github.com/AhsanSheraz/pyresilience.git
cd pyresilience
pip install -e ".[dev]"
```

--------------------------------

### Basic Pyresilience Installation with Poetry

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/getting-started/installation.md

Add Pyresilience to your project dependencies using Poetry.

```bash
poetry add pyresilience
```

--------------------------------

### Querying and Managing Registry Configurations

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/registry.md

Provides examples of how to retrieve specific configurations, fall back to defaults, list registered names, unregister configurations, and clear the entire registry.

```python
# Get a specific config
config = registry.get("payment-api")

# Get config with fallback to default
config = registry.get_or_default("unknown-api")

# List all registered names
names = registry.names  # ["payment-api", "inventory-api"]

# Remove a config
registry.unregister("payment-api")

# Clear everything
registry.clear()
```

--------------------------------

### Create and Attach Listeners

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Demonstrates how to create custom listeners and attach them to resilient functions using decorators. Includes examples for structured logging and metrics collection.

```python
logger = JsonEventLogger()
metrics = MetricsCollector()

def alert_on_circuit_open(event: ResilienceEvent) -> None:
    if event.event_type == EventType.CIRCUIT_OPEN:
        send_slack_alert(f"Circuit opened for {event.function_name}: {event.error}")
    elif event.event_type == EventType.CIRCUIT_CLOSED:
        send_slack_alert(f"Circuit recovered for {event.function_name}")

@resilient(
    retry=RetryConfig(max_attempts=3),
    circuit_breaker=CircuitBreakerConfig(failure_threshold=5),
    listeners=[logger, metrics, alert_on_circuit_open],
)
def payment_service(amount: float):
    return process_payment(amount)
```

```python
resilience_context.set({"trace_id": "abc-123", "user_id": "u-456"})
```

```python
summary = metrics.summary()
```

```python
counts = metrics.get_counts("payment_service")
```

```python
latencies = metrics.get_latencies("payment_service")
```

```python
metrics.reset()
```

```json
{"event_type": "retry", "function_name": "payment_service", "attempt": 1,
 "detail": "retrying in 1.00s", "error_type": "ConnectionError"}
{"event_type": "success", "function_name": "payment_service", "attempt": 2}
```

--------------------------------

### JsonEventLogger Output Example

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/observability.md

Example of the structured JSON output generated by `JsonEventLogger` for 'retry' and 'success' events.

```json
{"event_type": "retry", "function_name": "my_function", "attempt": 1, "detail": "retrying in 1.00s", "error_type": "ConnectionError", "error_message": "Connection refused"}
{"event_type": "success", "function_name": "my_function", "attempt": 2}
```

--------------------------------

### Rate Limiting with Waiting

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/ratelimiter.md

Configure rate limiting to allow callers to wait for a token instead of immediate rejection. This example allows 100 calls per minute and waits up to 5 seconds.

```python
# 100 calls per minute, wait up to 5 seconds for a token
@resilient(rate_limiter=RateLimiterConfig(
    max_calls=100,
    period=60.0,
    max_wait=5.0,
))
def call_api() -> dict:
    return requests.get("https://api.example.com").json()
```

--------------------------------

### Protect Connection Pools with Bulkheads

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/performance.md

Use the `BulkheadConfig` to protect database connection pools from being overwhelmed. This example limits the number of concurrent database queries to the size of the connection pool.

```python
@resilient(bulkhead=BulkheadConfig(max_concurrent=pool_size))
def db_query(sql: str):
    ...

```

--------------------------------

### MetricsCollector Summary Output

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/observability.md

Example of the summary output from `MetricsCollector`, including total events, counts per event type, success rate, and per-function statistics.

```python
{
    "total_events": 150,
    "event_counts": {
        "success": 120,
        "retry": 25,
        "failure": 5,
    },
    "success_rate": 0.8,
    "functions": {
        "my_function": {
            "total": 150,
            "success": 120,
            "failure": 5,
        }
    }
}
```

--------------------------------

### Async Support with Decorators

Source: https://github.com/ahsansheraz/pyresilience/blob/main/README.md

Use the resilient decorator with async functions without any changes. Ensure necessary libraries like aiohttp are installed for network operations.

```python
import aiohttp
from pyresilience import resilient, RetryConfig, CircuitBreakerConfig

@resilient(
    retry=RetryConfig(max_attempts=3, delay=0.5),
    circuit_breaker=CircuitBreakerConfig(failure_threshold=5),
)
async def call_api(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.json()
```

--------------------------------

### Configure Aggressive Caching for Read-Heavy Workloads

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/performance.md

Implement aggressive caching using `CacheConfig` to improve performance for read-heavy workloads. This example configures a cache with a maximum size of 1000 items and a time-to-live (TTL) of 60 seconds.

```python
@resilient(cache=CacheConfig(max_size=1000, ttl=60.0))
def get_config(key: str) -> str: ...

```

--------------------------------

### Configure Timeout for Function Execution

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Applies a timeout to function calls to prevent indefinite waiting. This example shows a basic timeout configuration and how to combine it with retries, specifying whether the timeout applies per-attempt or as a total deadline.

```python
from pyresilience import resilient, RetryConfig, TimeoutConfig
import requests

# Basic timeout - each attempt gets 5 seconds
@resilient(timeout=TimeoutConfig(seconds=5.0))
def slow_operation() -> dict:
    return requests.get("https://slow-api.example.com").json()

# Per-attempt timeout with retries (default behavior)
# Each retry attempt gets its own 10s timeout
@resilient(
    retry=RetryConfig(max_attempts=3, delay=1.0),
    timeout=TimeoutConfig(seconds=10, per_attempt=True),
)
def call_with_retry() -> dict:
    return requests.get("https://api.example.com").json()
# Total max time: 3 attempts * 10s + 2 delays = ~32s worst case

```

--------------------------------

### Basic Rate Limiting with Decorator

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/ratelimiter.md

Apply rate limiting to a function using the @resilient decorator. This example allows 10 requests per second.

```python
from pyresilience import resilient, RateLimiterConfig

@resilient(rate_limiter=RateLimiterConfig(max_calls=10, period=1.0))
def call_api(endpoint: str) -> dict:
    return requests.get(endpoint).json()
```

--------------------------------

### Handle Circuit Breaker Events

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/circuitbreaker.md

Implement event listeners to react to circuit breaker state changes. This example shows how to alert an operations team when the circuit opens.

```python
from pyresilience import resilient, CircuitBreakerConfig, EventType

def on_event(event):
    if event.event_type == EventType.CIRCUIT_OPEN:
        alert_ops_team(f"Circuit opened for {event.function_name}")

@resilient(
    circuit_breaker=CircuitBreakerConfig(failure_threshold=5),
    listeners=[on_event],
)
def critical_service():
    ... 
```

--------------------------------

### Apply Timeout with Retry Configuration

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/timelimiter.md

Combines retry logic with a 10-second timeout per attempt for API calls. Each retry gets its own timeout budget.

```python
@resilient(
    retry=RetryConfig(max_attempts=3, delay=1.0),
    timeout=TimeoutConfig(seconds=10),
)
def call_api() -> dict:
    return requests.get("https://api.example.com").json()
```

--------------------------------

### All Resilience Patterns Configuration

Source: https://github.com/ahsansheraz/pyresilience/blob/main/README.md

Configure all seven resilience patterns (retry, timeout, circuit breaker, fallback, bulkhead, rate limiter, cache) using a single @resilient decorator. This example shows comprehensive configuration options.

```python
from pyresilience import resilient, RetryConfig, TimeoutConfig, CircuitBreakerConfig
from pyresilience import FallbackConfig, BulkheadConfig, RateLimiterConfig, CacheConfig

@resilient(
    retry=RetryConfig(max_attempts=3, delay=1.0, backoff_factor=2.0),
    timeout=TimeoutConfig(seconds=10),
    circuit_breaker=CircuitBreakerConfig(failure_threshold=5, recovery_timeout=30),
    fallback=FallbackConfig(handler=lambda e: {"status": "degraded"}, fallback_on=[Exception]),
    bulkhead=BulkheadConfig(max_concurrent=10),
    rate_limiter=RateLimiterConfig(max_calls=100, period=60.0),
    cache=CacheConfig(ttl=300.0, max_size=1000),
)
def call_service(endpoint: str) -> dict:
    return requests.get(endpoint).json()
```

--------------------------------

### Observing Resilience Events with Listeners

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/index.md

Use listeners like JsonEventLogger, MetricsCollector, and custom functions to observe resilience events. The alert_on_circuit_open function provides an example of reacting to specific event types.

```python
from pyresilience import resilient, RetryConfig, CircuitBreakerConfig
from pyresilience import JsonEventLogger, MetricsCollector, EventType

logger = JsonEventLogger()
metrics = MetricsCollector()

def alert_on_circuit_open(event):
    if event.event_type == EventType.CIRCUIT_OPEN:
        print(f"ALERT: Circuit opened for {event.function_name}")

@resilient(
    retry=RetryConfig(max_attempts=3),
    circuit_breaker=CircuitBreakerConfig(failure_threshold=5),
    listeners=[logger, metrics, alert_on_circuit_open],
)
def payment_service(amount: float):
    return process_payment(amount)

# After some calls, check metrics:
# metrics.summary() -> {"total_events": 150, "success_rate": 0.95, ...}
```

--------------------------------

### Direct Cache Usage

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Shows direct interaction with the `ResultCache` class, including creating a cache instance, putting and getting entries, invalidating specific keys, and clearing the entire cache.

```python
cache = ResultCache(CacheConfig(max_size=100, ttl=60.0))
key = ResultCache.make_key("user", 42)
cache.put(key, {"name": "Alice"})
result = cache.get(key)  # {"name": "Alice"}
cache.invalidate(key)    # Remove specific entry
cache.clear()            # Clear all entries
```

--------------------------------

### Check for uvloop and orjson Availability

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/performance.md

Determine at runtime whether uvloop and orjson are installed and available for use with pyresilience. This allows for conditional logic based on the presence of these performance backends.

```python
from pyresilience import has_uvloop, has_orjson

print(f"uvloop: {has_uvloop()}")  # True/False
print(f"orjson: {has_orjson()}")  # True/False
```

--------------------------------

### Direct RateLimiter Usage

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/ratelimiter.md

Instantiate and use the RateLimiter directly without a decorator. The `acquire()` method attempts to get a token, returning True on success and False if the limit is exceeded. The `reset()` method clears all tokens.

```python
from pyresilience import RateLimiter, RateLimiterConfig

rl = RateLimiter(RateLimiterConfig(max_calls=10, period=1.0))

if rl.acquire():
    result = do_something()
else:
    print("Rate limit exceeded")

# Reset to full capacity
rl.reset()
```

--------------------------------

### Dynamic Registration at Application Startup

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/registry.md

Illustrates a pattern for registering resilience configurations dynamically during application startup, separating configuration logic from service definitions.

```python
# config.py
from pyresilience import ResilienceRegistry, ResilienceConfig, RetryConfig, CircuitBreakerConfig

registry = ResilienceRegistry()

def configure_resilience():
    """Call this at application startup."""
    registry.register("payment-api", ResilienceConfig(
        retry=RetryConfig(max_attempts=3),
        circuit_breaker=CircuitBreakerConfig(failure_threshold=5),
    ))
    registry.register("user-api", ResilienceConfig(
        retry=RetryConfig(max_attempts=2),
        timeout=TimeoutConfig(seconds=5),
    ))
```

```python
# services.py
from config import registry

@registry.decorator("payment-api")
async def charge(amount: float) -> dict: ...

@registry.decorator("user-api")
async def get_user(user_id: int) -> dict: ...
```

```python
# main.py
from config import configure_resilience

configure_resilience()
# Now all decorated functions use the registered configs
```

--------------------------------

### Basic Registry Usage with Multiple Configurations

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/registry.md

Shows how to initialize a ResilienceRegistry, register named configurations for different services, and decorate functions with these configurations.

```python
from pyresilience import ResilienceRegistry, ResilienceConfig, RetryConfig, CircuitBreakerConfig

registry = ResilienceRegistry()

# Register named configurations
registry.register("payment-api", ResilienceConfig(
    retry=RetryConfig(max_attempts=3),
    circuit_breaker=CircuitBreakerConfig(failure_threshold=5, recovery_timeout=30),
))

registry.register("inventory-api", ResilienceConfig(
    retry=RetryConfig(max_attempts=2),
    circuit_breaker=CircuitBreakerConfig(failure_threshold=3, recovery_timeout=60),
))

# Decorate functions
@registry.decorator("payment-api")
async def charge_card(amount: float) -> dict:
    return await payment_client.charge(amount)

@registry.decorator("payment-api")
async def refund_card(amount: float) -> dict:
    return await payment_client.refund(amount)

@registry.decorator("inventory-api")
async def check_stock(item_id: str) -> int:
    return await inventory_client.get_stock(item_id)
```

--------------------------------

### Get Resilience Health Status

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/observability.md

Use health_check function with a ResilienceRegistry to get the current state of resilience components.

```python
from pyresilience import ResilienceRegistry, health_check

registry = ResilienceRegistry()
# ... register configs and decorate functions ...

status = health_check(registry)
```

```python
@app.get("/health/resilience")
def resilience_health():
    return health_check(registry)
```

--------------------------------

### Built-in Presets for Common Use Cases

Source: https://github.com/ahsansheraz/pyresilience/blob/main/README.md

Leverage predefined policies for HTTP, database, queue, and strict latency-critical scenarios to quickly configure resilience.

```python
from pyresilience import resilient
from pyresilience import http_policy, db_policy, queue_policy, strict_policy

@resilient(**http_policy())       # 10s timeout, 3 retries, circuit breaker
def call_api(): ...

@resilient(**db_policy())         # 30s timeout, 2 retries, 10 concurrent max
def query_db(): ...

@resilient(**queue_policy())      # 15s timeout, 5 retries, high failure threshold
async def publish_message(): ...

@resilient(**strict_policy())     # 5s timeout, 1 retry, fail fast
def latency_critical(): ...
```

--------------------------------

### Get Per-Function Metrics from MetricsCollector

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/observability.md

Retrieve specific event counts for a particular function from the `MetricsCollector`.

```python
# Counts for a specific function
counts = metrics.get_counts("my_function")
```

--------------------------------

### Direct Rate Limiter Usage

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Demonstrates direct instantiation and usage of the `RateLimiter` class. Includes acquiring a token, handling `RateLimitExceededError`, and resetting the limiter.

```python
rl = RateLimiter(RateLimiterConfig(max_calls=10, period=1.0))
if rl.acquire():
    result = do_something()
else:
    print("Rate limit exceeded")
    time.sleep(1)

# Reset to full capacity
rl.reset()
```

--------------------------------

### Run Tests and Linting

Source: https://github.com/ahsansheraz/pyresilience/blob/main/CONTRIBUTING.md

Execute various testing and code quality checks using make commands. Use 'make test' for standard tests, 'make test-cov' for tests with coverage, 'make lint' for linting, and 'make typecheck' for type checking.

```bash
make test        # Run tests
make test-cov    # Run tests with coverage
make lint        # Run linting
make typecheck   # Run type checking
```

--------------------------------

### Apply Timeout to an Asynchronous Function

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/timelimiter.md

Applies a 5-second timeout to an asynchronous function using aiohttp for HTTP requests. Ensure aiohttp is installed.

```python
@resilient(timeout=TimeoutConfig(seconds=5.0))
async def async_fetch(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.json()
```

--------------------------------

### Apply Basic Timeout to a Synchronous Function

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/timelimiter.md

Applies a 5-second timeout to a synchronous function. Ensure the 'requests' library is installed if making HTTP calls.

```python
from pyresilience import resilient, TimeoutConfig

@resilient(timeout=TimeoutConfig(seconds=5.0))
def slow_operation() -> dict:
    return requests.get("https://slow-api.example.com").json()
```

--------------------------------

### Async Rate Limiting

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/ratelimiter.md

Implement rate limiting for asynchronous functions using the @resilient decorator. This example allows 50 calls per second.

```python
@resilient(rate_limiter=RateLimiterConfig(max_calls=50, period=1.0))
async def async_call() -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com") as resp:
            return await resp.json()
```

--------------------------------

### Using HTTP Policy Preset

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/getting-started/quickstart.md

Apply pre-defined resilience configurations for common use cases like HTTP requests using the http_policy preset.

```python
from pyresilience import resilient
from pyresilience.presets import http_policy

@resilient(**http_policy())
def call_api(url: str) -> dict:
    return requests.get(url).json()
```

--------------------------------

### Default Fallback Configuration

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/combining.md

Demonstrates the default fallback configuration where no specific exceptions are caught, effectively disabling fallback behavior.

```python
from pyresilience import resilient, FallbackConfig

# Safe — no fallback is actually triggered (fallback_on is empty)
@resilient(fallback=FallbackConfig())
def my_func(): ...

```

--------------------------------

### RateLimiter Configuration

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/ratelimiter.md

Configure the rate limiter with maximum calls, period in seconds, and maximum wait time for tokens. A max_wait of 0 rejects immediately.

```python
from pyresilience import RateLimiterConfig

config = RateLimiterConfig(
    max_calls=10,
    period=1.0,
    max_wait=0.0,
)
```

--------------------------------

### FastAPI Dependency Injection for Per-Route Resilience

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/frameworks.md

Use ResilientDependency with FastAPI's Depends() for fine-grained control over resilience for individual routes. Ensure pyresilience is installed.

```python
from fastapi import Depends, FastAPI
from pyresilience import ResilienceConfig, RetryConfig, CircuitBreakerConfig
from pyresilience.contrib.fastapi import ResilientDependency

app = FastAPI()

# Create a resilience dependency for the payment service
payment_resilience = ResilientDependency(ResilienceConfig(
    retry=RetryConfig(max_attempts=3),
    circuit_breaker=CircuitBreakerConfig(failure_threshold=5),
))

@app.post("/charge")
async def charge(
    amount: float,
    resilience: ResilientDependency = Depends(payment_resilience),
):
    return await resilience.call(payment_service.charge, amount)

@app.post("/refund")
async def refund(
    amount: float,
    resilience: ResilientDependency = Depends(payment_resilience),
):
    return await resilience.call(payment_service.refund, amount)
```

--------------------------------

### Using Database Policy Preset

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/getting-started/quickstart.md

Apply pre-defined resilience configurations for database operations using the db_policy preset.

```python
from pyresilience import resilient
from pyresilience.presets import http_policy, db_policy

@resilient(**db_policy())
def query_db(sql: str) -> list:
    return cursor.execute(sql).fetchall()
```

--------------------------------

### Configure Total Deadline Mode

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Use TimeoutConfig with seconds and per_attempt=False to set a total deadline across all retries. This ensures SLA compliance by bounding the overall execution time.

```python
from pyresilience import resilient, RetryConfig, TimeoutConfig

@resilient(
    retry=RetryConfig(max_attempts=3, delay=0.5),
    timeout=TimeoutConfig(seconds=30.0, per_attempt=False),
)
def strict_deadline_call() -> dict:
    """Bounded total latency for SLA compliance."""
    return requests.get("https://api.example.com").json()
```

--------------------------------

### Define a Custom Resilience Event Listener

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/observability.md

A custom listener is any callable that accepts a `ResilienceEvent`. This example defines a simple listener that prints event details to the console.

```python
from pyresilience import resilient, RetryConfig, ResilienceEvent

def my_listener(event: ResilienceEvent) -> None:
    print(f"[{event.event_type.value}] {event.function_name} "
          f"attempt={event.attempt} {event.detail}")

@resilient(
    retry=RetryConfig(max_attempts=3),
    listeners=[my_listener],
)
def my_function():
    return do_work()
```

--------------------------------

### Custom Policy Creation

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/presets.md

Users can create their own custom presets by defining a function that returns a dictionary of configuration parameters, allowing overrides of default retry, timeout, circuit breaker, and rate limiter settings.

```python
from pyresilience._types import RetryConfig, TimeoutConfig, CircuitBreakerConfig, RateLimiterConfig

def my_api_policy(**overrides):
    defaults = {
        "retry": RetryConfig(max_attempts=3, delay=0.5),
        "timeout": TimeoutConfig(seconds=15),
        "circuit_breaker": CircuitBreakerConfig(failure_threshold=5),
        "rate_limiter": RateLimiterConfig(max_calls=50, period=1.0),
    }
    defaults.update(overrides)
    return defaults
```

--------------------------------

### Run Tests with Coverage

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/contributing.md

Run tests and generate a coverage report. This command checks for branch coverage and fails if coverage is below 95%.

```bash
pytest --cov=pyresilience --cov-branch --cov-report=term-missing --cov-fail-under=95
```

--------------------------------

### Run Full Benchmark

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/getting-started/comparison.md

Executes the full performance benchmark suite for resilience libraries. This script is located in the benchmarks directory.

```python
python benchmarks/full_benchmark.py
```

--------------------------------

### Unified Resilience with @resilient() Decorator

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/index.md

Use the `@resilient()` decorator to combine multiple resilience patterns. This example configures retry, timeout, and circuit breaker for a payment API call.

```python
import requests
from pyresilience import resilient, RetryConfig, TimeoutConfig, CircuitBreakerConfig

@resilient(
    retry=RetryConfig(max_attempts=3, delay=1.0),
    timeout=TimeoutConfig(seconds=10),
    circuit_breaker=CircuitBreakerConfig(failure_threshold=5),
)
def call_payment_api(amount: float) -> dict:
    return requests.post("/charge", json={"amount": amount}).json()
```

--------------------------------

### Bulkhead with Waiting

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Configure a bulkhead to queue callers briefly, waiting for a slot to become available. Useful when temporary backlogs are acceptable.

```python
@resilient(bulkhead=BulkheadConfig(
    max_concurrent=5,
    max_wait=3.0,  # Wait up to 3 seconds for a slot
))
def call_api() -> dict:
    return requests.get("https://api.example.com").json()
```

--------------------------------

### Set Resilience Context in Web Framework Middleware

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/observability.md

Example of setting resilience context within web framework middleware to automatically include request-specific metadata in resilience events.

```python
@app.middleware("http")
async def add_resilience_context(request, call_next):
    resilience_context.set({
        "trace_id": request.headers.get("x-trace-id"),
        "path": request.url.path,
    })
    return await call_next(request)
```

--------------------------------

### Initialize JsonEventLogger and MetricsCollector

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Set up `JsonEventLogger` and `MetricsCollector` for monitoring resilience patterns. These components facilitate structured JSON logging and metrics collection.

```python
from pyresilience import (
    resilient,
    RetryConfig,
    CircuitBreakerConfig,
    JsonEventLogger,
    MetricsCollector,
    EventType,
    ResilienceEvent,
    resilience_context,
)
```

--------------------------------

### Bulkhead Exception Handling

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Demonstrates how to catch `BulkheadFullError` to handle cases where no concurrent slots are available and the maximum wait time is exceeded.

```python
try:
    result = protected_function()
except BulkheadFullError:
    # No concurrent slots available and max_wait exceeded
    return {"error": "Service busy, try again later"}
```

--------------------------------

### Track and Get In-Flight Call Count

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/api.md

Enable in-flight call tracking with `enable_in_flight_tracking()` to monitor the number of concurrently executing resilient calls. Use `get_in_flight_count()` to retrieve the current count.

```python
from pyresilience import enable_in_flight_tracking, get_in_flight_count

enable_in_flight_tracking()
# ... after some calls are in progress ...
count = get_in_flight_count()
```

--------------------------------

### Configure Timeout with Seconds and Per-Attempt

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/timelimiter.md

Defines a timeout configuration, specifying the maximum execution time in seconds and whether it applies per attempt or as a total deadline.

```python
from pyresilience import TimeoutConfig

config = TimeoutConfig(
    seconds=30.0,       # Maximum execution time
    per_attempt=True,    # True = per attempt, False = total deadline
)
```

--------------------------------

### Combining Multiple Resilience Patterns with @resilient()

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/index.md

This example demonstrates integrating retry, timeout, circuit breaker, fallback, rate limiter, and cache using the `@resilient()` decorator for an external API call.

```python
import requests
from pyresilience import resilient, RetryConfig, TimeoutConfig, CircuitBreakerConfig
from pyresilience import FallbackConfig, RateLimiterConfig, CacheConfig

@resilient(
    retry=RetryConfig(max_attempts=3, delay=1.0),
    timeout=TimeoutConfig(seconds=10),
    circuit_breaker=CircuitBreakerConfig(failure_threshold=5),
    fallback=FallbackConfig(handler=lambda e: {"error": "service unavailable"}),
    rate_limiter=RateLimiterConfig(max_calls=100, period=60.0),
    cache=CacheConfig(max_size=256, ttl=300.0),
)
def call_api():
    return requests.get("https://api.example.com/data").json()
# One decorator. All patterns. Shared state. Unified metrics.
```

--------------------------------

### Basic Bulkhead - Fail Immediately

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Implement a bulkhead pattern that rejects calls immediately if the maximum concurrent executions are reached. Useful for preventing resource exhaustion.

```python
@resilient(bulkhead=BulkheadConfig(
    max_concurrent=10,  # Max 10 concurrent executions
    max_wait=0.0,       # Fail immediately if no slot (default)
))
def query_database(sql: str) -> list:
    return cursor.execute(sql).fetchall()
```

--------------------------------

### Handling Sync and Async Functions with Registry

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/registry.md

Shows that the registry transparently handles both synchronous and asynchronous functions, applying the same named configuration to both.

```python
@registry.decorator("payment-api")
def sync_charge(amount: float) -> dict:
    return payment_client.charge(amount)

@registry.decorator("payment-api")
async def async_charge(amount: float) -> dict:
    return await payment_client.async_charge(amount)
```

--------------------------------

### Get Resilience Health Check Status

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/api.md

Obtain a summary of the resilience state for all registered functions by calling `health_check()` with a `ResilienceRegistry` instance. This provides insights into circuit breaker status, in-flight counts, and more.

```python
from pyresilience import ResilienceRegistry, health_check

registry = ResilienceRegistry()
status = health_check(registry)
# {"payment-api": {"circuit_breaker": "closed", "in_flight": 3, ...}, ...}
```

--------------------------------

### Basic CircuitBreakerConfig

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/circuitbreaker.md

Configure a circuit breaker to open after 5 consecutive failures, wait 30 seconds before attempting recovery, and require 2 successes in the half-open state to close. Specify which exceptions count as failures.

```python
from pyresilience import CircuitBreakerConfig

config = CircuitBreakerConfig(
    failure_threshold=5,      # Open after 5 consecutive failures
    recovery_timeout=30.0,    # Wait 30s before trying again
    success_threshold=2,      # Need 2 successes in half-open to close
    error_types=(Exception,), # Which exceptions count as failures
)
```

--------------------------------

### Async Circuit Breaker Usage

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/circuitbreaker.md

Apply a circuit breaker to an asynchronous function using the `@resilient` decorator. This example shows how to protect an async HTTP call with a circuit breaker configured to open after 5 failures.

```python
@resilient(circuit_breaker=CircuitBreakerConfig(failure_threshold=5))
async def async_call() -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com") as resp:
            return await resp.json()
```

--------------------------------

### Basic Circuit Breaker Usage

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/circuitbreaker.md

Apply a circuit breaker to a function using the `@resilient` decorator with basic configuration. This example protects `call_payment_service` which opens after 5 consecutive failures and waits 30 seconds before recovery.

```python
from pyresilience import resilient, CircuitBreakerConfig

@resilient(circuit_breaker=CircuitBreakerConfig(
    failure_threshold=5,
    recovery_timeout=30.0,
))
def call_payment_service(amount: float) -> dict:
    return requests.post("/charge", json={"amount": amount}).json()
```

--------------------------------

### Apply Built-in Database Policy

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Use the `db_policy` preset for database operations. It includes defaults for longer timeouts, fewer retries, and bulkhead protection.

```python
from pyresilience import resilient
from pyresilience.presets import db_policy

# Database calls - longer timeout, fewer retries, connection pool protection
# Defaults: 30s timeout, 2 retries, bulkhead of 10, circuit at 3 failures
@resilient(**db_policy())
def query_db(sql: str) -> list:
    return cursor.execute(sql).fetchall()
```

--------------------------------

### Implement Basic Circuit Breaker

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Use CircuitBreakerConfig to open a circuit after a specified number of consecutive failures. Configure recovery timeout and success threshold for transitioning back to closed state.

```python
from pyresilience import resilient, CircuitBreakerConfig, FallbackConfig, CircuitBreaker
import requests

# Basic circuit breaker - opens after 5 consecutive failures
@resilient(circuit_breaker=CircuitBreakerConfig(
    failure_threshold=5,       # Open after 5 consecutive failures
    recovery_timeout=30.0,     # Wait 30s before trying again (HALF_OPEN)
    success_threshold=2,       # Need 2 successes in HALF_OPEN to close
    error_types=(Exception,),  # Which exceptions count as failures
))
def call_payment_service(amount: float) -> dict:
    return requests.post("/charge", json={"amount": amount}).json()
```

--------------------------------

### Run Tests

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/contributing.md

Execute the test suite using pytest. This command runs all tests verbosely.

```bash
pytest -v
```

--------------------------------

### Configure Cache with Max Size and TTL

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/cache.md

Configure cache behavior including the maximum number of entries and their time-to-live. Use `ttl=0` for entries that never expire.

```python
from pyresilience import CacheConfig

config = CacheConfig(
    max_size=256,   # Maximum number of cached entries
    ttl=300.0,      # Time-to-live: 5 minutes
)
```

--------------------------------

### Initialize and Use MetricsCollector

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/observability.md

The `MetricsCollector` provides in-memory metrics for dashboards and health checks. Attach it as a listener to resilient functions to collect data.

```python
from pyresilience import resilient, RetryConfig, MetricsCollector

metrics = MetricsCollector()

@resilient(retry=RetryConfig(max_attempts=3), listeners=[metrics])
def my_function():
    return do_work()

# After some calls:
summary = metrics.summary()
```

--------------------------------

### Traditional Decorator Stacking (Without pyresilience)

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/index.md

Illustrates the complexity of combining individual resilience libraries (tenacity, pybreaker, wrapt_timeout_decorator) by stacking decorators, leading to uncoordinated state and configuration.

```python
# Three libraries, three configs, no coordination
from tenacity import retry, stop_after_attempt, wait_exponential
from pybreaker import CircuitBreaker
from wrapt_timeout_decorator import timeout

breaker = CircuitBreaker(fail_max=5)

@timeout(10)
@breaker
@retry(stop=stop_after_attempt(3), wait=wait_exponential())
def call_api():
    return requests.get("https://api.example.com/data").json()
# No fallback. No rate limiting. No caching. No shared metrics.
# The circuit breaker doesn't know about retries. Timeouts don't coordinate with backoff.
```

--------------------------------

### Configure Basic Retry with Exponential Backoff

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Sets up automatic retries for a function with exponential backoff. This is useful for handling transient network issues or temporary service unavailability. The `jitter` option helps prevent thundering herd problems.

```python
from pyresilience import resilient, RetryConfig
import requests

# Basic retry with exponential backoff
@resilient(retry=RetryConfig(
    max_attempts=5,           # 5 total attempts (1 initial + 4 retries)
    delay=1.0,                # 1 second initial delay
    backoff_factor=2.0,       # Double delay each retry: 1s, 2s, 4s, 8s
    max_delay=60.0,           # Cap delay at 60 seconds
    jitter=True,              # Add randomization to prevent thundering herd
))
def fetch_data(endpoint: str) -> dict:
    return requests.get(endpoint).json()

# Retry only on specific exceptions
@resilient(retry=RetryConfig(
    max_attempts=3,
    retry_on=(requests.ConnectionError, requests.Timeout),  # Only retry these
))
def call_api() -> dict:
    return requests.get("https://api.example.com").json()

# Retry based on response content (not just exceptions)
@resilient(retry=RetryConfig(
    max_attempts=5,
    delay=1.0,
    retry_on_result=lambda r: r.get("status") == 429,  # Retry on rate limit response
))
def rate_limited_api() -> dict:
    return requests.get("https://api.example.com/data").json()

# Result: fetch_data retries up to 5 times with delays: ~1s, ~2s, ~4s, ~8s (with jitter)

```

--------------------------------

### Full Resilience Configuration

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/getting-started/quickstart.md

Combine multiple resilience patterns including cache, circuit breaker, rate limiter, bulkhead, retry, timeout, and fallback for production-grade applications.

```python
from pyresilience import (
    resilient,
    RetryConfig,
    TimeoutConfig,
    CircuitBreakerConfig,
    FallbackConfig,
    BulkheadConfig,
    RateLimiterConfig,
    CacheConfig,
)

@resilient(
    retry=RetryConfig(max_attempts=3, delay=0.5, backoff_factor=2.0),
    timeout=TimeoutConfig(seconds=10),
    circuit_breaker=CircuitBreakerConfig(failure_threshold=5, recovery_timeout=30),
    fallback=FallbackConfig(handler=lambda e: {"error": str(e), "cached": True}),
    bulkhead=BulkheadConfig(max_concurrent=20),
    rate_limiter=RateLimiterConfig(max_calls=100, period=60.0),
    cache=CacheConfig(max_size=256, ttl=300.0),
)
def get_user_profile(user_id: int) -> dict:
    import requests
    return requests.get(f"https://api.example.com/users/{user_id}").json()
```

--------------------------------

### Rate Limiter Exception Handling

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Shows how to catch `RateLimitExceededError` to handle situations where the rate limit has been hit, typically by waiting before retrying.

```python
try:
    result = throttled_function()
except RateLimitExceededError:
    # Rate limit exceeded, try again later
    time.sleep(1)
```

--------------------------------

### Bulkhead with Fallback

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Combine bulkhead with a fallback handler to gracefully manage situations where the bulkhead is full. The fallback provides an alternative response.

```python
@resilient(
    bulkhead=BulkheadConfig(max_concurrent=10),
    fallback=FallbackConfig(
        handler=lambda e: {"status": "busy", "retry_after": 5},
        fallback_on=(BulkheadFullError,),
    ),
)
def get_data() -> dict:
    return requests.get("https://api.example.com/data").json()
```

--------------------------------

### Cache Combined with Other Patterns

Source: https://context7.com/ahsansheraz/pyresilience/llms.txt

Demonstrates integrating caching with other resilience patterns like retry and circuit breaker. The cache acts as the first layer, short-circuiting the entire pipeline if a valid entry exists.

```python
@resilient(
    cache=CacheConfig(max_size=256, ttl=300.0),
    retry=RetryConfig(max_attempts=3),
    circuit_breaker=CircuitBreakerConfig(failure_threshold=5),
)
def get_product(product_id: int) -> dict:
    return requests.get(f"https://api.example.com/products/{product_id}").json()
```

--------------------------------

### Separate Circuit Breakers Without Registry

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/advanced/registry.md

Demonstrates how functions decorated with identical circuit breaker configurations will have independent circuit breakers if not managed by a registry.

```python
# These have SEPARATE circuit breakers — one can be open while the other is closed
@resilient(circuit_breaker=CircuitBreakerConfig(failure_threshold=5))
def charge_card(): ...

@resilient(circuit_breaker=CircuitBreakerConfig(failure_threshold=5))
def refund_card(): ...
```

--------------------------------

### Configure Aggressive Backoff Strategy

Source: https://github.com/ahsansheraz/pyresilience/blob/main/docs/core/retry.md

Implement an aggressive backoff strategy with a small initial delay, a high backoff factor, and a maximum delay to cap the retry intervals.

```python
RetryConfig(delay=0.1, backoff_factor=3.0, max_delay=30.0)
# Delays: 0.1s, 0.3s, 0.9s, 2.7s, 8.1s, 24.3s, 30s (capped), ...
```

--------------------------------

### Per-Attempt Timeout Configuration

Source: https://github.com/ahsansheraz/pyresilience/blob/main/README.md

Configure timeouts to apply per individual attempt rather than a total deadline for a function call, using TimeoutConfig with per_attempt=True.

```python
from pyresilience import resilient, TimeoutConfig

@resilient(timeout=TimeoutConfig(seconds=5, per_attempt=True))   # 5s per attempt
def call_api(): ...

@resilient(timeout=TimeoutConfig(seconds=30, per_attempt=False))  # 30s total deadline
def call_db(): ...
```