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.

CircuitBreaker is the core class of PyBreaker. It wraps dangerous operations — typically integration points such as database calls or external API requests — with a component that can short-circuit calls when the downstream system is unhealthy. The breaker transitions between three states (closed, open, and half-open) based on observed failure and success counts.

​
Constructor

pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=60,
    success_threshold=1,
    exclude=None,
    listeners=None,
    state_storage=None,
    name=None,
    throw_new_error_on_trip=True,
)
​
fail_max
int
default:"5"
Maximum number of consecutive failures before the circuit opens. Once this threshold is reached the breaker transitions from closed to open and subsequent calls fail immediately.
​
reset_timeout
float
default:"60"
Number of seconds the circuit remains open before transitioning to half-open. After this period the breaker allows a single trial call through to check whether the downstream system has recovered.
​
success_threshold
int
default:"1"
Number of consecutive successes required while in half-open state before the circuit closes again and resumes normal operation.
​
exclude
Iterable[type | Callable]
Exceptions that should not count as failures. Each entry can be:
  • An exception type (or base class) — any exception that is a subclass of this type is excluded.
  • A callable that accepts the exception instance and returns True to exclude it.
Business-logic exceptions (e.g. ValueError, PermissionError) are typically good candidates for exclusion so they do not trip the breaker.
​
listeners
Sequence[CircuitBreakerListener]
One or more CircuitBreakerListener instances to register at creation time. Listeners receive callbacks for state changes, failures, successes, and pre-call events.
​
state_storage
CircuitBreakerStorage
Storage backend used to persist state and counters. Defaults to CircuitMemoryStorage(STATE_CLOSED) (in-process memory). Supply a CircuitRedisStorage instance to share state across multiple processes or hosts.
​
name
str
Human-readable identifier for this breaker, useful in log output and monitoring dashboards. Accessible at runtime via the name property.
​
throw_new_error_on_trip
bool
default:"true"
Controls the exception raised at the moment the circuit trips (i.e., when fail_max is reached):
  • True (default) — raises CircuitBreakerError when the circuit trips.
  • False — re-raises the original exception that caused the trip.
In both cases, while the circuit is open, subsequent calls always raise CircuitBreakerError.

​
Example

import pybreaker

db_breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=60,
    success_threshold=3,
    name='database',
    exclude=[ValueError],
    throw_new_error_on_trip=True,
)

​
Properties

​
fail_counter
int
Current number of consecutive failures recorded by the storage backend. Read-only — incremented automatically on each system error.
​
success_counter
int
Current number of consecutive successes recorded while the breaker is in half-open state. Read-only — reset automatically on state transitions.
​
fail_max
int
Get or set the failure threshold. Updating this property takes effect on the next call evaluation.
​
reset_timeout
float
Get or set the reset timeout in seconds. Updating this property takes effect the next time the open-state timeout is evaluated.
​
success_threshold
int
Get or set the number of consecutive successes required in half-open state before the circuit closes.
​
current_state
str
Current state string as reported directly by the storage backend — one of 'closed', 'open', or 'half-open'. Read-only. Always reflects the authoritative value from storage (important for shared backends such as Redis).
​
state
CircuitClosedState | CircuitOpenState | CircuitHalfOpenState
The cached state object. Automatically re-synced from current_state if the storage value has changed (e.g., updated by another process). Setting this property also notifies all registered listeners of the state change.
​
name
str | None
Get or set the human-readable name of this circuit breaker.
​
listeners
tuple[CircuitBreakerListener, ...]
Immutable tuple of currently registered listeners. Use the listener management methods below to add or remove listeners at runtime.
​
excluded_exceptions
tuple
Immutable tuple of currently registered exception exclusions (types and/or callables). Use the exception management methods below to modify this set at runtime.

​
Usage Methods

​
call(func, *args, **kwargs) → T

Call func with the supplied positional and keyword arguments, subject to the rules of the current circuit state. This method is thread-safe — it acquires an internal re-entrant lock before delegating to the state handler. 
result = db_breaker.call(query_db, user_id)
If the circuit is open and the reset timeout has not elapsed, call raises CircuitBreakerError immediately without invoking func.

​
call_async(func, *args, **kwargs)

Call an async Tornado coroutine func with the supplied arguments, subject to the rules of the current circuit state. Returns a Tornado coroutine — use yield or await in a Tornado context. Requires tornado to be installed; raises ImportError if it is not present. 
result = yield db_breaker.call_async(async_query, user_id)

​
calling() → ContextManager

Return a context manager that applies circuit breaker logic to the enclosed block. Equivalent to calling call() with an anonymous wrapper function. 
with db_breaker.calling():
    result = do_something()

​
State Management Methods

​
open() → bool

Force the circuit into the open state immediately, recording opened_at as the current UTC time. Returns the value of throw_new_error_on_trip so that callers know which exception style to use.

​
half_open() → None

Force the circuit into the half-open state. The next call will be allowed through as a trial; the result determines whether the circuit closes or re-opens.

​
close() → None

Force the circuit into the closed state and reset the success counter. Normal call execution resumes immediately.

​
Listener Methods

​
add_listener(listener)

Register a single CircuitBreakerListener with this breaker. Thread-safe.

​
add_listeners(*listeners)

Convenience method to register multiple listeners in one call. 
db_breaker.add_listeners(logging_listener, metrics_listener)

​
remove_listener(listener)

Unregister a previously registered listener. Thread-safe.

​
Exception Management Methods

​
add_excluded_exception(exception)

Add an exception type (or callable predicate) to the exclusion list. Excluded exceptions do not increment the failure counter. Thread-safe.

​
add_excluded_exceptions(*exceptions)

Convenience method to add multiple exclusions in one call. 
db_breaker.add_excluded_exceptions(ValueError, KeyError)

​
remove_excluded_exception(exception)

Remove a previously registered exclusion. Thread-safe.

​
is_system_error(exception) → bool

Return True if exception should be counted as a system failure (i.e., it is not in the exclusion list). Checks each registered exclusion:
  • If the exclusion is an exception type, uses issubclass to test.
  • If the exclusion is a callable, calls it with the exception and checks the return value.
if db_breaker.is_system_error(exc):
    # will be counted toward fail_max
    ...

​
__call__(*call_args, **call_kwargs) → Callable

CircuitBreaker instances are callable, which makes them usable directly as function decorators. Invoking the breaker object wraps the decorated function so that every call is routed through the circuit breaker’s state logic. Optionally accepts the keyword argument __pybreaker_call_async=True to wrap a Tornado coroutine instead of a regular function. 
@db_breaker
def my_func():
    pass
When called without arguments the breaker wraps the decorated function synchronously. For Tornado async coroutines pass __pybreaker_call_async=True: 
from tornado import gen

@db_breaker(__pybreaker_call_async=True)
@gen.coroutine
def my_async_func():
    pass
__pybreaker_call_async=True requires tornado to be installed. An ImportError is raised at decoration time if it is not available.

Build docs developers (and LLMs) love

Get started for freeTalk to us