pybreaker

Install the latest stable version of PyBreaker from PyPI using pip.

$ pip install pybreaker

A circuit breaker moves through three primary states:

1. closed (STATE_CLOSED): The normal state. Calls are allowed to pass through. If failures exceed fail_max, the circuit trips to open.
2. open (STATE_OPEN): Calls fail immediately with a CircuitBreakerError. The breaker stays in this state for the duration of reset_timeout.
3. half-open (STATE_HALF_OPEN): After the timeout, the breaker allows a trial call.
   • If the trial call succeeds (reaching success_threshold), the circuit returns to closed.
   • If the trial call fails, the circuit returns to open and the timeout resets.

You can guard functions using a decorator, the .call() method, or as a context manager.

# 1. Using a decorator
@db_breaker
def update_customer(cust):
    pass

# 2. Using the .call() method
def update_customer(cust):
    pass

db_breaker.call(update_customer, my_customer)

# 3. Using a context manager
with db_breaker.calling():
    # Do stuff here...
    pass

PyBreaker supports asynchronous Tornado functions. Use the __pybreaker_call_async=True argument in the decorator or use the .call_async() method.

from tornado import gen
import pybreaker

# Using decorator
@db_breaker(__pybreaker_call_async=True)
@gen.coroutine
def async_update(cust):
    yield gen.sleep(1)

# Using .call_async()
@gen.coroutine
def async_update(cust):
    yield gen.sleep(1)

result = yield db_breaker.call_async(async_update, my_customer)

By default, any exception trips the circuit. You can exclude business exceptions (which don't indicate system instability) by passing exception types or callables to the exclude parameter.

import pybreaker

# Exclude by type
db_breaker = pybreaker.CircuitBreaker(exclude=[CustomerValidationError])

# Exclude using a callable (e.g., checking status codes)
db_breaker = pybreaker.CircuitBreaker(
    exclude=[lambda e: type(e) == HTTPError and e.status_code < 500]
)

# Add excluded exceptions later
db_breaker.add_excluded_exception(CustomerValidationError)

To react to circuit breaker events (like state changes or failures) without subclassing CircuitBreaker, subclass CircuitBreakerListener and implement the desired hooks. You can add listeners at creation time or dynamically later.

import pybreaker

class DBListener(pybreaker.CircuitBreakerListener):
    def before_call(self, cb, func, *args, **kwargs):
        pass

    def state_change(self, cb, old_state, new_state):
        pass

    def failure(self, cb, exc):
        pass

    def success(self, cb):
        pass

# Add at creation
db_breaker = pybreaker.CircuitBreaker(listeners=[DBListener()])

# Add later
db_breaker.add_listeners(DBListener())

You can inspect and manually control the state of a CircuitBreaker instance using its properties and methods.

# Monitoring
print(db_breaker.fail_counter)      # Current consecutive failures
print(db_breaker.success_counter)   # Current consecutive successes
print(db_breaker.fail_max)           # Max failures allowed
print(db_breaker.success_threshold)  # Successes needed to close
print(db_breaker.reset_timeout)      # Current timeout period
print(db_breaker.current_state)     # 'open', 'half-open', or 'closed'

# Management
db_breaker.close()                  # Force close the circuit
db_breaker.half_open()               # Force half-open state
db_breaker.open()                   # Force open the circuit

# Update configuration
db_breaker.fail_max = 10
db_breaker.success_threshold = 3
db_breaker.reset_timeout = 60

By default, pybreaker uses in-memory storage. To share circuit breaker state across multiple processes or servers, use CircuitRedisStorage. This requires the redis package to be installed.

Parameters for CircuitRedisStorage:

• state: Initial state.
• redis_object: A redis.Redis instance.
• namespace: (Optional) A prefix for the Redis keys.
• fallback_circuit_state: The state to use if Redis is unavailable (defaults to STATE_CLOSED).
• cluster_mode: (Boolean) Enables atomic updates for opened_at using Redis transactions.

import redis
import pybreaker

r = redis.Redis()
storage = pybreaker.CircuitRedisStorage(state='closed', redis_object=r, namespace='my_app')
breaker = pybreaker.CircuitBreaker(state_storage=storage)

Extend CircuitBreakerListener to hook into the lifecycle of a circuit breaker. Override these methods to perform actions like logging, alerting, or metrics collection.

Available Hooks:

• before_call(cb, func, *args, **kwargs): Called before the guarded function is executed.
• failure(cb, exc): Called when the guarded function fails.
• success(cb): Called when the guarded function succeeds.
• state_change(cb, old_state, new_state): Called when the circuit transitions between states.

import pybreaker

class MyListener(pybreaker.CircuitBreakerListener):
    def state_change(self, cb, old_state, new_state):
        print(f"Circuit {cb.name} changed from {old_state} to {new_state}")

    def failure(self, cb, exc):
        print(f"Call failed with: {exc}")

breaker = pybreaker.CircuitBreaker(listeners=[MyListener()])

You can execute guarded functions using three different patterns:

1. Direct Call: Pass the function and its arguments to .call().
2. Context Manager: Use the with breaker.calling(): block.
3. Decorator: Use the breaker instance as a decorator on a function definition.

For asynchronous functions (Tornado coroutines), use .call_async() or pass __pybreaker_call_async=True to the decorator.

# 1. Direct Call
result = breaker.call(my_function, arg1, arg2)

# 2. Context Manager
with breaker.calling():
    my_function(arg1, arg2)

# 3. Decorator
@breaker
def my_function(arg1):
    return arg1

# Async (Tornado)
@breaker( __pybreaker_call_async=True )
def my_async_function():
    yield