aiobreaker-redis

🛡 Distributed async circuit breaker for Python asyncio, httpx, and FastAPI with Redis state synchronization.

💡 Features

• ⚡️ Pure asyncio First: Zero blocking I/O, built natively for Python async/await applications.
• 🌐 Distributed Redis State: Synchronize circuit breaker status (CLOSED, OPEN, HALF_OPEN) across multiple Gunicorn/Uvicorn worker processes and Kubernetes pods.
• 🎯 Fast Fallbacks: Execute sync or async fallbacks instantly when circuits trip open.
• 🚀 FastAPI & httpx Ready: Clean @breaker.wrap(fallback=...) decorator for API endpoints and HTTP client wrappers.

📦 Installation

pip install aiobreaker-redis

🚀 Quick Usage

import httpx
from aiobreaker_redis import CircuitBreaker, CircuitOpenError

breaker = CircuitBreaker(
    name="payment-gateway",
    failure_threshold=5,
    recovery_timeout=30.0,
    # Optional redis client:
    # redis_client=redis_client
)

async def payment_fallback():
    return {"status": "queued", "message": "Payment service temporarily unavailable"}

@breaker.wrap(fallback=payment_fallback)
async def charge_card(amount: float):
    async with httpx.AsyncClient() as client:
        response = await client.post("https://api.payments.com/charge", json={"amount": amount})
        response.raise_for_status()
        return response.json()

📄 License

MIT © Raj Dubey