{"id":8814,"library":"aiobreaker","title":"aiobreaker: Circuit Breaker for asyncio","description":"aiobreaker is a Python implementation of the Circuit Breaker pattern, specifically designed for asynchronous applications using `asyncio`. It helps improve the resilience of microservices by preventing an application from repeatedly trying to execute an operation that is likely to fail, such as calling a service that is temporarily down. The current version is 1.2.0, with a stable, infrequent release cadence reflecting its focused scope.","status":"active","version":"1.2.0","language":"en","source_language":"en","source_url":"https://github.com/arlyon/aiobreaker","tags":["asyncio","circuit-breaker","resilience","fault-tolerance","microservices"],"install":[{"cmd":"pip install aiobreaker","lang":"bash","label":"Install stable version"}],"dependencies":[],"imports":[{"symbol":"CircuitBreaker","correct":"from aiobreaker import CircuitBreaker"},{"symbol":"CircuitBreakerError","correct":"from aiobreaker import CircuitBreakerError"},{"symbol":"CircuitBreakerMonitor","correct":"from aiobreaker import CircuitBreakerMonitor"},{"symbol":"CircuitBreakerState","correct":"from aiobreaker.state import CircuitBreakerState"}],"quickstart":{"code":"import asyncio\nfrom aiobreaker import CircuitBreaker, CircuitBreakerError\n\n# Configure a Circuit Breaker: open after 3 failures, stay open for 5 seconds\nbreaker = CircuitBreaker(fail_max=3, timeout_duration=5)\n\n# Simulate an unreliable asynchronous service\nfailure_count = 0\n\n@breaker\nasync def unreliable_service():\n    global failure_count\n    if failure_count < 4: # Intentionally fail 4 times to trip the breaker (fail_max=3)\n        failure_count += 1\n        print(f\"Service attempt failed ({failure_count}/{breaker.fail_max})\")\n        raise ConnectionError(\"Simulated network issue\")\n    print(\"Service successful (after breaker resets/trips)\")\n    return \"Data retrieved\"\n\nasync def main():\n    print(\"\\n--- Starting service calls ---\")\n    for i in range(12):\n        print(f\"\\nCall {i+1}:\")\n        try:\n            result = await unreliable_service()\n            print(f\"Call successful: {result}\")\n        except CircuitBreakerError:\n            print(\"CircuitBreaker is OPEN! Blocking service calls to prevent further load.\")\n        except ConnectionError as e:\n            print(f\"Call failed with transient error: {e}\")\n        await asyncio.sleep(1)\n\n    print(\"\\n--- All calls attempted ---\")\n\nif __name__ == \"__main__\":\n    asyncio.run(main())\n","lang":"python","description":"Demonstrates how to apply a `CircuitBreaker` to an `async` function that simulates failures. It shows how the breaker opens after `fail_max` attempts and how to catch `CircuitBreakerError` to implement fallback logic or notify users of service unavailability."},"warnings":[{"fix":"Always wrap calls to `@breaker` decorated functions in a `try...except CircuitBreakerError` block to implement fallback logic or notify users gracefully.","message":"Failing to handle `aiobreaker.errors.CircuitBreakerError` can lead to unhandled exceptions when the circuit breaker opens. The purpose of a circuit breaker is to allow for graceful degradation, not just to block calls.","severity":"gotcha","affected_versions":"All"},{"fix":"Ensure that any function decorated with `@breaker` is an `async def` function and is always called using `await`.","message":"`aiobreaker` is designed exclusively for `asyncio` applications. Using it with synchronous functions will lead to unexpected behavior or runtime errors, as it expects awaitable callables.","severity":"gotcha","affected_versions":"All"},{"fix":"Carefully consider the typical latency, error rates, and recovery times of your target service. Start with conservative values and adjust based on monitoring and testing in a production-like environment.","message":"Incorrectly configuring `fail_max` or `timeout_duration` can render the circuit breaker ineffective. Too low `fail_max` might trip it unnecessarily; too high might not protect against failing services. Incorrect `timeout_duration` might keep the service unavailable too long or try too soon.","severity":"gotcha","affected_versions":"All"},{"fix":"Use `exclude` only for exceptions that you are certain are transient and should not contribute to tripping the breaker (e.g., specific HTTP 4xx errors that indicate client-side issues, not service unavailability). Ensure that common service-failure exceptions (e.g., `ConnectionError`, `TimeoutError`, HTTP 5xx) are *not* excluded.","message":"The `exclude` parameter in `CircuitBreaker` allows specifying exceptions that *should not* trip the breaker. Misusing this (e.g., excluding critical errors or including transient ones) can lead to the breaker either never opening or opening too aggressively.","severity":"gotcha","affected_versions":"All"}],"env_vars":null,"last_verified":"2026-04-16T00:00:00.000Z","next_check":"2026-07-15T00:00:00.000Z","problems":[{"fix":"This is expected behavior. Implement a fallback mechanism or error handling in your `except CircuitBreakerError:` block. Monitor the underlying service to understand why it is failing.","cause":"The decorated asynchronous function failed a configured number of times (`fail_max`), causing the circuit breaker to transition to an OPEN state, blocking further calls.","error":"aiobreaker.errors.CircuitBreakerError: CircuitBreaker is open"},{"fix":"Ensure the function decorated with `@breaker` is defined as `async def` and always invoke it with `await function_name(...)`.","cause":"You are attempting to call a function decorated with `@breaker` without `await`ing it, or the decorated function itself is not an `async def`.","error":"TypeError: 'coroutine' object is not callable"},{"fix":"Add `from aiobreaker import CircuitBreaker` at the top of your script or module.","cause":"The `CircuitBreaker` class was not correctly imported into your Python file.","error":"NameError: name 'CircuitBreaker' is not defined"}]}