or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

async.mdcircuit-breaker.mdindex.mdlisteners.mdstorage.md
tile.json

tessl/pypi-pybreaker

Python implementation of the Circuit Breaker pattern for handling failing subsystems gracefully

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/pybreaker@1.4.x

To install, run

npx @tessl/cli install tessl/pypi-pybreaker@1.4.0

index.mddocs/

PyBreaker

A robust Python implementation of the Circuit Breaker design pattern, allowing applications to handle failing subsystems gracefully without cascading failures. The library offers configurable failure thresholds and reset timeouts, supports event listeners for monitoring circuit state changes, can guard both synchronous and asynchronous (Tornado) functions, and provides thread-safe operation with optional Redis backing for distributed systems.

Package Information

  • Package Name: pybreaker
  • Language: Python
  • Installation: pip install pybreaker
  • Python Version: 3.9+

Core Imports

import pybreaker

Common usage pattern:

from pybreaker import CircuitBreaker, CircuitBreakerListener

Feature availability can be checked:

# Check if Redis support is available
print(pybreaker.HAS_REDIS_SUPPORT)  # True if redis package installed

# Check if Tornado async support is available  
print(pybreaker.HAS_TORNADO_SUPPORT)  # True if tornado package installed

Basic Usage

import pybreaker

# Create a circuit breaker with default settings
db_breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=60)

# Use as decorator
@db_breaker
def database_call():
    # Potentially failing operation
    # Will be protected by circuit breaker
    pass

# Use with direct call
def another_db_call():
    # Some database operation
    pass

result = db_breaker.call(another_db_call)

# Use as context manager
with db_breaker.calling():
    # Protected code block
    pass

Architecture

PyBreaker implements the Circuit Breaker pattern with three core states:

  • Closed State: Normal operation, calls pass through to the wrapped function
  • Open State: Circuit is open, calls fail immediately with CircuitBreakerError
  • Half-Open State: Testing state allowing one trial call to determine if circuit should close

The architecture supports:

  • Storage Backends: Memory-based and Redis-based storage for circuit state
  • Event System: Listener pattern for monitoring circuit events and state changes
  • Thread Safety: All operations are thread-safe using internal locking
  • Exception Filtering: Configurable exclusion of business exceptions from circuit logic

Capabilities

Core Circuit Breaker

Main circuit breaker functionality with configurable failure thresholds, reset timeouts, and state management. Supports decorator, direct call, and context manager usage patterns.

class CircuitBreaker:
    def __init__(self, fail_max: int = 5, reset_timeout: float = 60, 
                 success_threshold: int = 1, 
                 exclude: Iterable[type | Callable[[Any], bool]] | None = None, 
                 listeners: Sequence[CircuitBreakerListener] | None = None, 
                 state_storage: CircuitBreakerStorage | None = None, 
                 name: str | None = None, 
                 throw_new_error_on_trip: bool = True): ...
    
    def call(self, func, *args, **kwargs): ...
    def calling(self): ...  # Context manager
    def open(self) -> bool: ...
    def close(self) -> None: ...
    def half_open(self) -> None: ...

Core Circuit Breaker

Storage Backends

Storage implementations for persisting circuit breaker state across application restarts and distributed systems. Includes in-memory storage for single-process applications and Redis storage for distributed scenarios.

class CircuitBreakerStorage:
    def __init__(self, name: str) -> None: ...
    # Abstract base class with state management methods

class CircuitMemoryStorage:
    def __init__(self, state: str): ...

class CircuitRedisStorage:
    def __init__(self, state: str, redis_object: Redis, namespace: str | None = None, 
                 fallback_circuit_state: str = "closed", cluster_mode: bool = False): ...

Storage Backends

Event Listeners

Event listener system for monitoring circuit breaker state changes, failures, and successes. Enables integration with logging, metrics, and alerting systems.

class CircuitBreakerListener:
    def before_call(self, cb: CircuitBreaker, func: Callable[..., Any], *args: Any, **kwargs: Any) -> None: ...
    def failure(self, cb: CircuitBreaker, exc: BaseException) -> None: ...
    def success(self, cb: CircuitBreaker) -> None: ...
    def state_change(self, cb: CircuitBreaker, old_state: CircuitBreakerState | None, new_state: CircuitBreakerState) -> None: ...

Event Listeners

Async Support

Optional Tornado integration for protecting asynchronous operations with circuit breaker functionality. Requires tornado package to be installed.

def call_async(self, func, *args, **kwargs): ...

Async Support

State Constants

STATE_OPEN = "open"
STATE_CLOSED = "closed"  
STATE_HALF_OPEN = "half-open"

# Feature availability flags
HAS_REDIS_SUPPORT: bool  # True if redis package is available
HAS_TORNADO_SUPPORT: bool  # True if tornado package is available

Exception Types

class CircuitBreakerError(Exception):
    """Raised when circuit breaker is open and calls are rejected."""