DOCSEARCHbyAlgolia
Submit Docs

fabfuel/circuitbreaker

repository·develop·Indexed 19 days ago

https://github.com/fabfuel/circuitbreaker

A Python implementation of the Circuit Breaker pattern designed to prevent cascading failures in distributed systems. It provides a @circuit decorator for synchronous and asynchronous functions to monitor execution and temporarily halt calls to failing integration points based on configurable failure thresholds and recovery timeouts.

Tokens
1.2K
Snippets
5
Records
6
Agent score
18%

What's inside fabfuel-circuitbreaker

  1. Use the @circuit decorator for basic circuit breaking

    develop

    The simplest way to use the library is to decorate a function (synchronous or asynchronous) with @circuit. This sets up a circuit breaker with default settings:

    • Failure threshold: 5 subsequent failures.
    • Recovery timeout: 30 seconds.
    • Expected exception: Any exception inheriting from Exception.
    • Name: The name of the decorated function.

    When the circuit is open, subsequent calls will raise a CircuitBreaker exception.

    from circuitbreaker import circuit

@circuit
def external_call():
    ...

@circuit
async def async_external_call():
    ...

  3. Customize failure logic with a callable

    develop

    If simple type checking is insufficient, pass a callable to expected_exception. The callable receives (thrown_type, thrown_value) and must return True if the exception should be treated as a failure. This is useful for inspecting exception attributes, such as HTTP status codes.

    # Assume we are using the requests library
def is_not_http_error(thrown_type, thrown_value):
    return issubclass(thrown_type, RequestException) and not issubclass(thrown_type, HTTPError)

def is_rate_limited(thrown_type, thrown_value):
    return issubclass(thrown_type, HTTPError) and thrown_value.status_code == 429

@circuit(expected_exception=is_not_http_error)
def call_flaky_api(...):
    rsp = requests.get(...)
    rsp.raise_for_status()
    return rsp

@circuit(expected_exception=is_rate_limited)
def call_slow_server(...):
    rsp = requests.get(...)
    rsp.raise_for_status()
    return rsp

  4. Configure circuit breaker parameters

    develop

    You can customize the behavior of the circuit breaker by passing arguments to the @circuit decorator:

    • failure_threshold: Number of subsequent failures before the circuit opens (default: 5).
    • recovery_timeout: Seconds to wait in the 'open' state before entering 'half-open' (default: 30).
    • expected_exception: The exception(s) that trigger a failure. Can be an exception class, an iterable of classes, or a callable (default: Exception).
    • name: A custom name for the circuit breaker (default: function name).
    • fallback_function: A function to call instead of raising CircuitBreaker when the circuit is open. The fallback must match the signature and type (sync/async) of the decorated function.
    from circuitbreaker import circuit

@circuit(failure_threshold=10, expected_exception=ConnectionError)
def external_call():
    ...

  5. Create a custom CircuitBreaker subclass

    develop

    For reusable configurations, extend the CircuitBreaker class. You can then apply it using the class instance as a decorator or via the cls parameter in the @circuit proxy.

    from circuitbreaker import CircuitBreaker, circuit

class MyCircuitBreaker(CircuitBreaker):
    FAILURE_THRESHOLD = 10
    RECOVERY_TIMEOUT = 60
    EXPECTED_EXCEPTION = RequestException

# Option 1: Use as an object (must be initialized with parentheses)
@MyCircuitBreaker()
def external_call():
    ...

# Option 2: Use via the decorator proxy
@circuit(cls=MyCircuitBreaker)
def external_call():
    ...

  6. Monitor circuit breaker states

    develop

    Every circuit breaker automatically registers itself with the CircuitBreakerMonitor. You can use the monitor to inspect the health of your system:

    • CircuitBreakerMonitor.get_circuits(): Returns all registered circuit breakers.
    • CircuitBreakerMonitor.all_closed(): Returns True if all circuits are closed.
    • CircuitBreakerMonitor.get_open(): Returns currently open circuits.
    • CircuitBreakerMonitor.get_closed(): Returns currently closed circuits.