circuitbreaker 1.0.1

CircuitBreaker

This is an Python implementation of the “Circuit Breaker” Pattern (http://martinfowler.com/bliki/CircuitBreaker.html). Inspired by Michael T. Nygard’s highly recommendable book Release It! (https://pragprog.com/book/mnee/release-it).

Installation

The project is available on PyPI. Simply run:

$ pip install circuitbreaker

Usage

This is the simplest example. Just decorate a function with the @circuit decorator:

from circuitbreaker import circuit

@circuit
def external_call():
    ...

This decorator sets up a circuit breaker with the default settings. The circuit breaker:

• monitors the function execution and counts failures

• resets the failure count after every successful execution (while is is closed)

• opens and prevents further executions after 5 subsequent failures

• switches to half-open and allows one test-exection after 30 seconds recovery timeout

• closes if the test-execution succeeded

• considers all raised exceptions (based on class Exception) as an expected failure

• is named “external_call” - the name of the function it decorates

Configuration

The following configuration options can be adjusted via decorator parameters. For example:

from circuitbreaker import circuit

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

Advanced Usage

If you apply circuit breakers to a couple of functions and you always set specific options other than the default values, you can extend the CircuitBreaker class and create your own circuit breaker subclass instead:

from circuitbreaker import CircuitBreaker

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

Now you have two options to apply your circuit breaker to a function. As an Object directly:

@MyCircuitBreaker()
def external_call():
    ...

Please note, that the circuit breaker class has to be initialized, you have to use a class instance as decorator (@MyCircuitBreaker()), not the class itself (@MyCircuitBreaker).

Or via the decorator proxy:

@circuit(cls=MyCircuitBreaker)
def external_call():
    ...

Monitoring

To keep track of the health of your application and the state of your circuit breakers, every circuit breaker registers itself at the CircuitBreakerMonitor. You can receive all registered circuit breakers via CircuitBreakerMonitor.get_circuits().

To get an aggregated health status, you can ask the Monitor via CircuitBreakerMonitor.all_closed(). Or you can retrieve the currently open circuits via CircuitBreakerMonitor.get_open() and the closed circuits via CircuitBreakerMonitor.get_closed().

Todo

• add unit tests