Registry / http-networking / aiobreaker

aiobreaker

JSON →
library1.2.0pypypi✓ verified 85d ago

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.

pip install aiobreaker
INSTALL
IMPORT
SIG · AIOBREAKER
A
aiobreaker
http-networkingpythonv1.2.0
Install
1.6s avg
Import
210ms
Disk
16MB
Pass rate
10/ 10
Env Coverage10 / 10
glibc
3.93.13
musl
3.93.13
Install & Compatibility
Where this runs
tested against v1.2.0 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
musl
py 3.103.920 runs
installs and imports cleanly · install 0.0s · import 0.222s · 18MB
glibc
py 3.103.920 runs
installs and imports cleanly · install 1.6s · import 0.198s · 19MB
16MB installed
● package 16MB
Code
Verified usage

Verified import paths — ran on the pinned version, not inferred.

CircuitBreaker
from aiobreaker import CircuitBreaker
CircuitBreakerError
from aiobreaker import CircuitBreakerError
CircuitBreakerMonitor
from aiobreaker import CircuitBreakerMonitor
CircuitBreakerState
from aiobreaker.state import CircuitBreakerState

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.

import asyncio from aiobreaker import CircuitBreaker, CircuitBreakerError # Configure a Circuit Breaker: open after 3 failures, stay open for 5 seconds breaker = CircuitBreaker(fail_max=3, timeout_duration=5) # Simulate an unreliable asynchronous service failure_count = 0 @breaker async def unreliable_service(): global failure_count if failure_count < 4: # Intentionally fail 4 times to trip the breaker (fail_max=3) failure_count += 1 print(f"Service attempt failed ({failure_count}/{breaker.fail_max})") raise ConnectionError("Simulated network issue") print("Service successful (after breaker resets/trips)") return "Data retrieved" async def main(): print("\n--- Starting service calls ---") for i in range(12): print(f"\nCall {i+1}:") try: result = await unreliable_service() print(f"Call successful: {result}") except CircuitBreakerError: print("CircuitBreaker is OPEN! Blocking service calls to prevent further load.") except ConnectionError as e: print(f"Call failed with transient error: {e}") await asyncio.sleep(1) print("\n--- All calls attempted ---") if __name__ == "__main__": asyncio.run(main())
Debug
Known issues
gotchaFailing 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.
fix
Always wrap calls to `@breaker` decorated functions in a `try...except CircuitBreakerError` block to implement fallback logic or notify users gracefully.
affects: All
gotcha`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.
fix
Ensure that any function decorated with `@breaker` is an `async def` function and is always called using `await`.
affects: All
gotchaIncorrectly 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.
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.
affects: All
gotchaThe `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.
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.
affects: All
Errors
Common errors & fixes
aiobreaker.errors.CircuitBreakerError: CircuitBreaker is open
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.
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.
TypeError: 'coroutine' object is not callable
You are attempting to call a function decorated with `@breaker` without `await`ing it, or the decorated function itself is not an `async def`.
fix
Ensure the function decorated with `@breaker` is defined as `async def` and always invoke it with `await function_name(...)`.
NameError: name 'CircuitBreaker' is not defined
The `CircuitBreaker` class was not correctly imported into your Python file.
fix
Add `from aiobreaker import CircuitBreaker` at the top of your script or module.
Upgrade
Version history
1.2.0latest on PyPI · released May 17, 2021
Audit
Dependencies

No dependency data recorded yet.

Agent activity
57 hits · last 30 days
node
48
OpenAI (training)
1
Resources