Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/danielfm/pybreaker/llms.txt

Use this file to discover all available pages before exploring further.

PyBreaker’s listener system lets you hook into circuit breaker lifecycle events without modifying CircuitBreaker itself. Attach one or more CircuitBreakerListener subclasses to receive callbacks whenever a call is attempted, succeeds, fails, or causes a state transition. This is the recommended way to integrate circuit breaker telemetry with logging frameworks, metrics systems, and alerting pipelines.

​
The CircuitBreakerListener Interface

Subclass CircuitBreakerListener and override only the methods you need. The base class provides no-op implementations for all four callbacks, so partial implementations are fully supported. 
class CircuitBreakerListener:
    def before_call(self, cb, func, *args, **kwargs):
        """Called before the circuit breaker calls func."""

    def failure(self, cb, exc):
        """Called when a function invocation raises a system error."""

    def success(self, cb):
        """Called when a function invocation succeeds."""

    def state_change(self, cb, old_state, new_state):
        """Called when the circuit breaker state changes."""
CallbackWhen it firesKey parameters
before_callImmediately before each protected callcb — the breaker; func — the callable; forwarded args/kwargs
failureWhen a call raises a system error (not an excluded exception)cb — the breaker; exc — the exception instance
successWhen a protected call returns without raisingcb — the breaker
state_changeWhen the circuit transitions between statescb — the breaker; old_state — previous state object; new_state — new state object

​
Example: Logging Listener

A logging listener is the simplest useful implementation — it records state changes and call outcomes to the standard Python logging system. 
import logging
import pybreaker

class LogListener(pybreaker.CircuitBreakerListener):
    """Logs circuit breaker state changes."""

    def state_change(self, cb, old_state, new_state):
        msg = "State Change: CB: {0}, New State: {1}".format(cb.name, new_state)
        logging.info(msg)

    def failure(self, cb, exc):
        logging.warning("Circuit breaker '%s' recorded failure: %s", cb.name, exc)

    def success(self, cb):
        logging.debug("Circuit breaker '%s' call succeeded", cb.name)

​
Example: Metrics Listener

A metrics listener forwards circuit breaker events to an external monitoring system. The example below targets StatsD, but the same pattern applies to Prometheus counters, DataDog custom metrics, or any other backend. 
class MetricsListener(pybreaker.CircuitBreakerListener):
    """Reports circuit breaker metrics to StatsD."""

    def __init__(self, statsd_client):
        self.statsd = statsd_client

    def failure(self, cb, exc):
        self.statsd.increment(f"circuit_breaker.{cb.name}.failure")

    def success(self, cb):
        self.statsd.increment(f"circuit_breaker.{cb.name}.success")

    def state_change(self, cb, old_state, new_state):
        self.statsd.event(
            f"Circuit breaker state change",
            f"{cb.name}: {old_state} -> {new_state}"
        )

​
Registering Listeners

You can register listeners at circuit breaker creation time, or add them later after the breaker is already in use. 
# At creation time
db_breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=60,
    listeners=[LogListener(), MetricsListener(statsd)]
)

# Or later
db_breaker.add_listener(LogListener())
db_breaker.add_listeners(LogListener(), MetricsListener(statsd))

​
Removing Listeners

Hold a reference to a listener instance to remove it later. You can also inspect all currently registered listeners via the listeners property. 
log_listener = LogListener()
db_breaker.add_listener(log_listener)

# Later...
db_breaker.remove_listener(log_listener)

# Inspect current listeners
print(db_breaker.listeners)  # tuple of registered listeners
Multiple listeners are fully supported. PyBreaker calls all registered listeners in the order they were added. Each listener receives the same event independently, so a failure in one listener’s callback does not prevent the others from being notified.

Build docs developers (and LLMs) love

Get started for freeTalk to us