aiobreaker: Circuit Breaker for asyncio

Common errors

• aiobreaker.errors.CircuitBreakerError: CircuitBreaker is open

  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.
  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

  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`.
  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

  cause 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.

Warnings

• gotcha 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.
  Fix: Always wrap calls to `@breaker` decorated functions in a `try...except CircuitBreakerError` block to implement fallback logic or notify users gracefully.
• 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`.
• gotcha 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.
  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.
• gotcha 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.
  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.

Install

• pip install aiobreaker Install stable version

Imports

• CircuitBreaker

  from aiobreaker import CircuitBreaker

• CircuitBreakerError

  from aiobreaker import CircuitBreakerError

• CircuitBreakerMonitor

  from aiobreaker import CircuitBreakerMonitor

• CircuitBreakerState

  from aiobreaker.state import CircuitBreakerState

Quickstart

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())