ss-utils-circuit-breaker 0.0.1

Async circuit breaker with optional Redis (coredis) support for distributed state

ss-utils-circuit-breaker

A Python async circuit breaker implementation with optional Redis (coredis) support for distributed state.

Installation

# Basic installation (in-memory state only)
uv add ss-utils-circuit-breaker

# With Redis support
uv add ss-utils-circuit-breaker[redis]

Quick Start

Basic Usage (In-Memory)

from ss_utils_circuit_breaker import CircuitBreaker, CircuitBreakerError

# Create a circuit breaker
breaker = CircuitBreaker(
    name="my_api",
    failure_threshold=5,  # Open after 5 failures
    recovery_timeout=60,  # Try again after 60 seconds
)

@breaker
async def call_external_api():
    # Your async API call here
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        response.raise_for_status()
        return response.json()

# Usage
try:
    data = await call_external_api()
except CircuitBreakerError:
    # Circuit is open - service is known to be unavailable
    data = get_cached_data()
except Exception:
    # Regular failure
    pass

With Redis (Distributed State)

from ss_utils_circuit_breaker import RedisCircuitBreaker

# Create a Redis-backed circuit breaker
breaker = RedisCircuitBreaker(
    name="my_api",
    redis_url="localhost:6379",
    namespace="circuit_breaker",  # Redis key prefix
    failure_threshold=5,
    recovery_timeout=60,
)

@breaker
async def call_external_api():
    # State is shared across all instances via Redis
    pass

Circuit Breaker States

CLOSED (Normal)
    │
    ▼ (failures >= threshold)
   OPEN (Blocking)
    │
    ▼ (after recovery_timeout)
HALF_OPEN (Testing)
    │
    ├─► (success) ──► CLOSED
    │
    └─► (failure) ──► OPEN

• CLOSED: Normal operation. Calls pass through, failures are counted.
• OPEN: Circuit is tripped. Calls fail immediately with CircuitBreakerError.
• HALF_OPEN: After timeout, one test call is allowed. Success closes circuit, failure reopens it.

Configuration Example

import httpx
from ss_utils_circuit_breaker import CircuitBreaker

# Only count connection errors as failures
api_breaker = CircuitBreaker(
    name="external_api",
    failure_threshold=3,
    recovery_timeout=30,
    expected_exception=(httpx.ConnectError, httpx.TimeoutException),
)

@api_breaker
async def fetch_data():
    async with httpx.AsyncClient(timeout=10.0) as client:
        return await client.get("https://api.example.com/data")

Testing

# Run tests
uv run pytest

# With coverage
uv run pytest --cov

License

MIT