### Async Call Example with PyBreaker

Source: https://context7.com/danielfm/pybreaker/llms.txt

Demonstrates how to use PyBreaker with asynchronous functions. It shows both decorator and call_async usage, including handling IOError and CircuitBreakerError.

```python
import pybreaker
from tornado import gen, ioloop

@gen.coroutine
def async_save(data):
    raise IOError("storage unavailable")

@gen.coroutine
def async_fetch(url):
    # This is a placeholder for an actual async fetch operation
    yield gen.sleep(0.1) # Simulate network latency
    return {'url': url, 'status': 200}

breaker = pybreaker.CircuitBreaker()

@gen.coroutine
def main():
    # Decorator usage
    result = yield async_fetch("https://api.example.com/data")
    print(result)  # {'url': 'https://api.example.com/data', 'status': 200}

    # call_async usage
    try:
        yield breaker.call_async(async_save, {"key": "value"})
    except IOError:
        print(f"Async call failed, counter={breaker.fail_counter}")
    except pybreaker.CircuitBreakerError:
        print("Circuit open — skipping async call")

ioloop.IOLoop.current().run_sync(main)
```

--------------------------------

### Monitoring and Observability with PyBreaker

Source: https://context7.com/danielfm/pybreaker/llms.txt

Shows how to inspect and modify PyBreaker's runtime properties like state, counters, and thresholds. Includes an example of a health-check endpoint.

```python
import pybreaker

breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=60, name="inventory")

# Read current state and counters
print(breaker.current_state)    # 'closed'
print(breaker.fail_counter)     # current consecutive failure count
print(breaker.success_counter)  # consecutive successes in half-open state
print(breaker.name)             # 'inventory'

# Adjust thresholds at runtime without recreating the breaker
breaker.fail_max = 10
breaker.reset_timeout = 120
breaker.success_threshold = 3
breaker.name = "inventory_v2"

# Health-check endpoint example (Flask-style pseudocode)
def health():
    return {
        "circuit": breaker.name,
        "state": breaker.current_state,
        "fail_counter": breaker.fail_counter,
        "fail_max": breaker.fail_max,
        "reset_timeout": breaker.reset_timeout,
        "success_threshold": breaker.success_threshold,
    }

print(health())
# {'circuit': 'inventory_v2', 'state': 'closed', 'fail_counter': 0,
#  'fail_max': 10, 'reset_timeout': 120, 'success_threshold': 3}
```

--------------------------------

### List Available Commands

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Display a list of all available commands for the project's command-line interface.

```bash
$ ./pw -i
```

--------------------------------

### Run Tests

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Execute the project's tests using the provided command.

```bash
$ ./pw test
```

--------------------------------

### Configure CircuitMemoryStorage

Source: https://context7.com/danielfm/pybreaker/llms.txt

Use `CircuitMemoryStorage` for in-process state management. This is the default and suitable for single-instance services.

```python
import pybreaker

# Explicit in-memory storage (this is also the default when state_storage is omitted)
storage = pybreaker.CircuitMemoryStorage(pybreaker.STATE_CLOSED)
breaker = pybreaker.CircuitBreaker(fail_max=3, state_storage=storage)

```

--------------------------------

### Pre-initialize Circuit Breaker State

Source: https://context7.com/danielfm/pybreaker/llms.txt

Useful for setting up a circuit breaker in a specific state, particularly in testing scenarios. Demonstrates state constants.

```python
open_storage = pybreaker.CircuitMemoryStorage(pybreaker.STATE_OPEN)
pre_opened_breaker = pybreaker.CircuitBreaker(state_storage=open_storage)
print(pre_opened_breaker.current_state)  # 'open'

# State constants
print(pybreaker.STATE_CLOSED)    # 'closed'
print(pybreaker.STATE_OPEN)      # 'open'
print(pybreaker.STATE_HALF_OPEN) # 'half-open'
```

--------------------------------

### Format and Lint Code

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Apply code formatting with black and isort, and perform linting with mypy using the provided commands.

```bash
$ ./pw format
```

```bash
$ ./pw lint
```

--------------------------------

### Initialize CircuitBreaker with Django Redis and Default Connection

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Use an existing Redis connection from django_redis for the CircuitBreaker's state storage.

```python
import pybreaker
from django_redis import get_redis_connection

db_breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=60,
    state_storage=pybreaker.CircuitRedisStorage(pybreaker.STATE_CLOSED, get_redis_connection('default')))
```

--------------------------------

### Initialize CircuitBreaker for Database Integration

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Create a CircuitBreaker instance for database operations with a failure threshold of 5 and a reset timeout of 60 seconds.

```python
import pybreaker

# Used in database integration points
db_breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=60)
```

--------------------------------

### Create CircuitBreaker Instance

Source: https://context7.com/danielfm/pybreaker/llms.txt

Instantiate a CircuitBreaker with default or custom configurations. Parameters include fail_max, reset_timeout, success_threshold, exception filters, and a name. Instances should be created at application scope.

```python
import pybreaker

# Minimal — defaults: fail_max=5, reset_timeout=60s, success_threshold=1
db_breaker = pybreaker.CircuitBreaker()

# Fully configured
db_breaker = pybreaker.CircuitBreaker(
    fail_max=3,            # open after 3 consecutive failures
    reset_timeout=30,      # wait 30 s before allowing a trial call
    success_threshold=2,   # require 2 successes in half-open before closing
    exclude=[ValueError],  # ValueError is a business error, not a system error
    name="db_breaker",     # name used in logs/monitoring
    throw_new_error_on_trip=True,  # raise CircuitBreakerError when circuit trips
)

print(db_breaker.current_state)   # 'closed'
print(db_breaker.fail_max)        # 3
print(db_breaker.reset_timeout)   # 30
print(db_breaker.success_threshold)  # 2
```

--------------------------------

### Manual State Control - open(), half_open(), close()

Source: https://context7.com/danielfm/pybreaker/llms.txt

Manually control the circuit breaker's state at runtime using `open()`, `half_open()`, and `close()` methods. This is useful for maintenance windows, health checks, or administrative dashboards.

```APIDOC
## Manual State Control — open(), half_open(), close()

Allows operators to manually override the circuit state at runtime, useful for maintenance windows, health-check endpoints, or administrative dashboards.

### Methods
- **open()**: Forces the circuit breaker into the 'open' state.
- **half_open()**: Transitions the circuit breaker to the 'half-open' state.
- **close()**: Forces the circuit breaker into the 'closed' state and resets the failure counter.

### Runtime Reconfiguration
Circuit breaker properties like `fail_max`, `reset_timeout`, and `success_threshold` can be modified at runtime.

### Usage
```python
import pybreaker

breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=30)

# Force open during a planned maintenance window
breaker.open()
print(breaker.current_state)  # 'open'

# Put into half-open to probe recovery
breaker.half_open()
print(breaker.current_state)  # 'half-open'

# Force close when the system is confirmed healthy
breaker.close()
print(breaker.current_state)  # 'closed'
print(breaker.fail_counter)   # 0  (reset on close)

# Runtime reconfiguration
breaker.fail_max = 10
breaker.reset_timeout = 120
breaker.success_threshold = 3
print(breaker.fail_max)          # 10
print(breaker.reset_timeout)     # 120
print(breaker.success_threshold) # 3
```
```

--------------------------------

### Initialize CircuitBreaker with Unique Redis Namespace

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Configure a CircuitBreaker with Redis storage, specifying a unique namespace to prevent state conflicts when using multiple independent circuit breakers.

```python
import pybreaker
from django_redis import get_redis_connection

db_breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=60,
    state_storage=pybreaker.CircuitRedisStorage(pybreaker.STATE_CLOSED, get_redis_connection('default'),namespace='unique_namespace'))
```

--------------------------------

### Configure CircuitBreaker with Success Threshold

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Initialize a CircuitBreaker that requires 3 successful requests before closing the circuit, in addition to the failure threshold and reset timeout.

```python
# Require 3 successful requests before closing
db_breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=60, success_threshold=3)
```

--------------------------------

### Initialize CircuitBreaker with Redis Storage

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Configure a CircuitBreaker to use Redis for state storage. Ensure the Redis connection is not initialized with decode_responses=True.

```python
import pybreaker
import redis

redis = redis.StrictRedis()
db_breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=60,
    state_storage=pybreaker.CircuitRedisStorage(pybreaker.STATE_CLOSED, redis))
```

--------------------------------

### CircuitBreaker Constructor

Source: https://context7.com/danielfm/pybreaker/llms.txt

Creates a circuit breaker instance. All parameters are optional. Instances should be created at application scope and reused across requests.

```APIDOC
## CircuitBreaker — Constructor

Creates a circuit breaker instance. All parameters are optional. Instances should be created at application scope (module-level or as singletons) and reused across requests.

```python
import pybreaker

# Minimal — defaults: fail_max=5, reset_timeout=60s, success_threshold=1
db_breaker = pybreaker.CircuitBreaker()

# Fully configured
db_breaker = pybreaker.CircuitBreaker(
    fail_max=3,            # open after 3 consecutive failures
    reset_timeout=30,      # wait 30 s before allowing a trial call
    success_threshold=2,   # require 2 successes in half-open before closing
    exclude=[ValueError],  # ValueError is a business error, not a system error
    name="db_breaker",     # name used in logs/monitoring
    throw_new_error_on_trip=True,  # raise CircuitBreakerError when circuit trips
)

print(db_breaker.current_state)   # 'closed'
print(db_breaker.fail_max)        # 3
print(db_breaker.reset_timeout)   # 30
print(db_breaker.success_threshold)  # 2
```
```

--------------------------------

### CircuitBreaker.calling() - Context Manager

Source: https://context7.com/danielfm/pybreaker/llms.txt

Use the `calling()` method as a context manager to guard arbitrary blocks of code, especially useful for third-party code or lambdas where direct decoration is not possible.

```APIDOC
## CircuitBreaker.calling() — Context Manager

Returns a context manager so a circuit breaker can guard an arbitrary block of code with a `with` statement. Useful when you cannot decorate the function directly (e.g., third-party code or lambdas).

### Usage
```python
import pybreaker

breaker = pybreaker.CircuitBreaker(fail_max=2, reset_timeout=5)

def process_order(order_id):
    try:
        with breaker.calling():
            # Guarded block — any exception here is counted as a failure
            result = risky_inventory_check(order_id)
            return result
    except pybreaker.CircuitBreakerError:
        # Return a safe fallback when the circuit is open
        return {"status": "unavailable", "order_id": order_id}
    except Exception as exc:
        # Real error — circuit failure was already counted
        raise

# After 2 failures the circuit opens
print(breaker.current_state)  # eventually 'open'
```
```

--------------------------------

### Use Circuit Breaker with Asynchronous Tornado Functions (call_async)

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Alternatively, use the call_async method to wrap asynchronous Tornado function calls without decorator syntax.

```python
@gen.coroutine
def async_update(cust):
    # Do async stuff here...
    pass

updated_customer = db_breaker.call_async(async_update, my_customer)
```

--------------------------------

### Configure Circuit Breaker to Throw New Error on Trip

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Set throw_new_error_on_trip to False to allow the original error to be thrown when the circuit trips.

```python
pybreaker.CircuitBreaker(..., throw_new_error_on_trip=False)
```

--------------------------------

### Tornado Async Support

Source: https://context7.com/danielfm/pybreaker/llms.txt

Wraps asynchronous Tornado coroutines using the decorator with `__pybreaker_call_async=True` or by using `call_async()`. Requires the 'tornado' package.

```python
import pybreaker
from tornado import gen, ioloop

breaker = pybreaker.CircuitBreaker(fail_max=3, reset_timeout=10)

# Decorator style
@breaker(__pybreaker_call_async=True)
@gen.coroutine
def async_fetch(url):
    """Async HTTP fetch guarded by the circuit breaker."""
    # ... actual async HTTP call ...
    raise gen.Return({"url": url, "status": 200})

```

--------------------------------

### Use Circuit Breaker with call() Method

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Alternatively, use the call() method to wrap a function call without using decorator syntax.

```python
def update_customer(cust):
    # Do stuff here...
    pass

# Will trigger the circuit breaker
updated_customer = db_breaker.call(update_customer, my_customer)
```

--------------------------------

### Redis-Backed Circuit Breaker

Source: https://context7.com/danielfm/pybreaker/llms.txt

Implements a distributed circuit breaker using Redis for shared state across multiple application instances. Requires the 'redis' package. Note: do not use decode_responses=True for the Redis client.

```python
import redis
import pybreaker

redis_client = redis.StrictRedis(host="localhost", port=6379, db=0)

# Basic Redis-backed circuit breaker
storage = pybreaker.CircuitRedisStorage(
    state=pybreaker.STATE_CLOSED,
    redis_object=redis_client,
    namespace="payments",            # isolates keys: "payments:pybreaker:state" etc.
    fallback_circuit_state=pybreaker.STATE_CLOSED,  # state to use if Redis is unavailable
)
breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=60, state_storage=storage)

# Redis cluster mode (uses SET instead of WATCH/MULTI/EXEC transactions)
cluster_storage = pybreaker.CircuitRedisStorage(
    state=pybreaker.STATE_CLOSED,
    redis_object=redis_client,
    namespace="inventory",
    cluster_mode=True,
)
cluster_breaker = pybreaker.CircuitBreaker(state_storage=cluster_storage)

# Using django-redis
from django_redis import get_redis_connection
django_storage = pybreaker.CircuitRedisStorage(
    pybreaker.STATE_CLOSED,
    get_redis_connection("default"),
    namespace="order_service",
)
```

--------------------------------

### Direct Invocation with CircuitBreaker.call()

Source: https://context7.com/danielfm/pybreaker/llms.txt

Use the .call() method to execute a synchronous function through the circuit breaker. This method raises CircuitBreakerError when the circuit is open. It tracks failures and opens the circuit after exceeding fail_max.

```python
import pybreaker

breaker = pybreaker.CircuitBreaker(fail_max=3, reset_timeout=10)

def fetch_user(user_id):
    # Simulates a database call
    if user_id < 0:
        raise ConnectionError("DB unreachable")
    return {"id": user_id, "name": "Alice"}

# Successful call — fail_counter stays 0
result = breaker.call(fetch_user, 42)
print(result)           # {'id': 42, 'name': 'Alice'}
print(breaker.fail_counter)   # 0
print(breaker.current_state)  # 'closed'

# After 3 failures the circuit opens
for _ in range(2):
    try:
        breaker.call(fetch_user, -1)
    except ConnectionError:
        pass

try:
    breaker.call(fetch_user, -1)
except pybreaker.CircuitBreakerError as e:
    print(f"Circuit opened: {e}")   # 'Failures threshold reached, circuit breaker opened'

print(breaker.current_state)   # 'open'

# Subsequent calls fail immediately without touching the DB
try:
    breaker.call(fetch_user, 42)
except pybreaker.CircuitBreakerError as e:
    print(f"Still open: {e}")   # 'Timeout not elapsed yet, circuit breaker still open'
```

--------------------------------

### Use Circuit Breaker with Context Manager

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

The circuit breaker can also be used as a context manager with a 'with' statement for a block of code.

```python
# Will trigger the circuit breaker
with db_breaker.calling():
    # Do stuff here...
    pass
```

--------------------------------

### CircuitMemoryStorage - In-Process State

Source: https://context7.com/danielfm/pybreaker/llms.txt

The default state storage backend for Pybreaker. `CircuitMemoryStorage` keeps state and counters in memory within a single process, suitable for single-instance services.

```APIDOC
## CircuitMemoryStorage — In-Process State

The default storage backend. State and counters are kept in memory within a single process. Suitable for single-instance services; not shared across processes or machines.

### Usage
```python
import pybreaker

# Explicit in-memory storage (this is also the default when state_storage is omitted)
storage = pybreaker.CircuitMemoryStorage(pybreaker.STATE_CLOSED)
breaker = pybreaker.CircuitBreaker(fail_max=3, state_storage=storage)
```
```

--------------------------------

### Controlled Recovery with Success Threshold

Source: https://context7.com/danielfm/pybreaker/llms.txt

Configures a circuit breaker to require a specific number of consecutive successful calls in the half-open state before fully closing. This prevents premature recovery from flapping services.

```python
import pybreaker
from time import sleep

# Require 3 successes in half-open before closing
breaker = pybreaker.CircuitBreaker(
    fail_max=3,
    reset_timeout=0.1,
    success_threshold=3,
)

def unstable_service():
    raise ConnectionError("down")

def healthy_service():
    return "ok"

# Trip the circuit
for _ in range(2):
    try: breaker.call(unstable_service) 
    except ConnectionError: pass
try: breaker.call(unstable_service)
except pybreaker.CircuitBreakerError: pass

print(breaker.current_state)   # 'open'
sleep(0.15)  # wait for reset_timeout

# First success — still half-open
breaker.call(healthy_service)
print(breaker.current_state)    # 'half-open'
print(breaker.success_counter)  # 1

# Second success — still half-open
breaker.call(healthy_service)
print(breaker.current_state)    # 'half-open'
print(breaker.success_counter)  # 2

# Third success — circuit closes
breaker.call(healthy_service)
print(breaker.current_state)    # 'closed'
print(breaker.success_counter)  # 0  (reset)
```

--------------------------------

### CircuitBreakerListener - Event Hooks

Source: https://context7.com/danielfm/pybreaker/llms.txt

Extend the `CircuitBreakerListener` class to implement custom logic for logging, metrics, alerting, or distributed tracing. Listeners can be added or removed at runtime.

```APIDOC
## CircuitBreakerListener — Event Hooks

Subclass `CircuitBreakerListener` to plug in custom logic for logging, metrics, alerting, or distributed tracing without modifying the core circuit breaker.

### Listener Methods
- **before_call(cb, func, *args, **kwargs)**: Called before a guarded function is invoked.
- **success(cb)**: Called when a guarded function call succeeds.
- **failure(cb, exc)**: Called when a guarded function call raises an exception.
- **state_change(cb, old_state, new_state)**: Called when the circuit breaker's state changes.

### Runtime Listener Management
Listeners can be added to or removed from a circuit breaker instance at runtime using `add_listener()` and `remove_listener()`.

### Usage
```python
import logging
import pybreaker

logger = logging.getLogger(__name__)

class ObservabilityListener(pybreaker.CircuitBreakerListener):
    def before_call(self, cb, func, *args, **kwargs):
        logger.debug("CB '%s' calling %s", cb.name, func.__name__)

    def success(self, cb):
        logger.info("CB '%s' call succeeded (fails=%d)", cb.name, cb.fail_counter)

    def failure(self, cb, exc):
        logger.warning("CB '%s' call FAILED [%s] (fails=%d/%d)",
                       cb.name, type(exc).__name__, cb.fail_counter, cb.fail_max)

    def state_change(self, cb, old_state, new_state):
        logger.error("CB '%s' state: %s -> %s",
                     cb.name,
                     old_state.name if old_state else "none",
                     new_state.name)

breaker = pybreaker.CircuitBreaker(
    fail_max=3,
    name="payments",
    listeners=[ObservabilityListener()],
)

# Add / remove listeners at runtime
extra = pybreaker.CircuitBreakerListener()
breaker.add_listener(extra)
breaker.remove_listener(extra)
print(len(breaker.listeners))  # 1
```
```

--------------------------------

### Monitor Circuit Breaker State and Counters

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Access properties like fail_counter, success_counter, fail_max, success_threshold, and reset_timeout to monitor the circuit breaker's status. The current_state property shows if the circuit is 'open', 'half-open', or 'closed'.

```python
# Get the current number of consecutive failures
print(db_breaker.fail_counter)
```

```python
# Get the current number of consecutive successes
print(db_breaker.success_counter)
```

```python
# Get/set the maximum number of consecutive failures
print(db_breaker.fail_max)
db_breaker.fail_max = 10
```

```python
# Get/set the success threshold
print(db_breaker.success_threshold)
db_breaker.success_threshold = 3
```

```python
# Get/set the current reset timeout period (in seconds)
print db_breaker.reset_timeout
db_breaker.reset_timeout = 60
```

```python
# Get the current state, i.e., 'open', 'half-open', 'closed'
print(db_breaker.current_state)
```

--------------------------------

### Exclude Exceptions from Circuit Breaker Failures

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Configure the circuit breaker to ignore specific exceptions, treating them as non-failures. This can be done at creation time or later.

```python
# At creation time...
db_breaker = CircuitBreaker(exclude=[CustomerValidationError])
```

```python
# ...or later
db_breaker.add_excluded_exception(CustomerValidationError)
```

```python
db_breaker = CircuitBreaker(exclude=[lambda e: type(e) == HTTPError and e.status_code < 500])
```

--------------------------------

### Use CircuitBreaker as a Context Manager

Source: https://context7.com/danielfm/pybreaker/llms.txt

Guard arbitrary blocks of code with a `with` statement when direct function decoration is not possible. Catches `CircuitBreakerError` to provide a fallback.

```python
import pybreaker

breaker = pybreaker.CircuitBreaker(fail_max=2, reset_timeout=5)

def process_order(order_id):
    try:
        with breaker.calling():
            # Guarded block — any exception here is counted as a failure
            result = risky_inventory_check(order_id)
            return result
    except pybreaker.CircuitBreakerError:
        # Return a safe fallback when the circuit is open
        return {"status": "unavailable", "order_id": order_id}
    except Exception as exc:
        # Real error — circuit failure was already counted
        raise

# After 2 failures the circuit opens
print(breaker.current_state)  # eventually 'open'
```

--------------------------------

### Define Logging CircuitBreaker Listener

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Implement a listener to log circuit breaker state changes, useful for monitoring and debugging.

```python
class LogListener(pybreaker.CircuitBreakerListener):
    """Listener used to log circuit breaker events."""

    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)
```

--------------------------------

### Add Listeners to Circuit Breaker

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Listeners can be added to a circuit breaker either at creation time or later using the add_listeners method.

```python
db_breaker = pybreaker.CircuitBreaker(listeners=[DBListener(), LogListener()])
```

```python
db_breaker.add_listeners(OneListener(), AnotherListener())
```

--------------------------------

### CircuitBreaker.call() - Direct Invocation

Source: https://context7.com/danielfm/pybreaker/llms.txt

Calls a regular (synchronous) function through the circuit breaker. Raises `CircuitBreakerError` when the circuit is open; otherwise propagates exceptions from the wrapped function normally.

```APIDOC
## CircuitBreaker.call() — Direct Invocation

Calls a regular (synchronous) function through the circuit breaker. Raises `CircuitBreakerError` when the circuit is open; otherwise propagates exceptions from the wrapped function normally (or as `CircuitBreakerError` when the circuit first trips, depending on `throw_new_error_on_trip`).

```python
import pybreaker

breaker = pybreaker.CircuitBreaker(fail_max=3, reset_timeout=10)

def fetch_user(user_id):
    # Simulates a database call
    if user_id < 0:
        raise ConnectionError("DB unreachable")
    return {"id": user_id, "name": "Alice"}

# Successful call — fail_counter stays 0
result = breaker.call(fetch_user, 42)
print(result)           # {'id': 42, 'name': 'Alice'}
print(breaker.fail_counter)   # 0
print(breaker.current_state)  # 'closed'

# After 3 failures the circuit opens
for _ in range(2):
    try:
        breaker.call(fetch_user, -1)
    except ConnectionError:
        pass

try:
    breaker.call(fetch_user, -1)
except pybreaker.CircuitBreakerError as e:
    print(f"Circuit opened: {e}")   # 'Failures threshold reached, circuit breaker opened'

print(breaker.current_state)   # 'open'

# Subsequent calls fail immediately without touching the DB
try:
    breaker.call(fetch_user, 42)
except pybreaker.CircuitBreakerError as e:
    print(f"Still open: {e}")   # 'Timeout not elapsed yet, circuit breaker still open'
```
```

--------------------------------

### CircuitBreaker as a Decorator

Source: https://context7.com/danielfm/pybreaker/llms.txt

Using a `CircuitBreaker` instance as a function decorator is the most common pattern. `functools.wraps` preserves the decorated function's `__name__` and `__doc__`.

```APIDOC
## CircuitBreaker as a Decorator

Using a `CircuitBreaker` instance as a function decorator is the most common pattern. `functools.wraps` preserves the decorated function's `__name__` and `__doc__`.

```python
import pybreaker

api_breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=60)

@api_breaker
def get_payment_status(payment_id: str) -> dict:
    """Fetch payment status from the payment gateway."""
    # ... real HTTP call here ...
    raise TimeoutError("Gateway timeout")

try:
    status = get_payment_status("pay_123")
except TimeoutError:
    print("Call failed, failure counted")
except pybreaker.CircuitBreakerError:
    print("Circuit is open — gateway is down, skipping call")

print(get_payment_status.__name__)  # 'get_payment_status'
print(get_payment_status.__doc__)   # 'Fetch payment status from the payment gateway.'
print(api_breaker.fail_counter)     # 1
```
```

--------------------------------

### Manually Control Circuit Breaker State

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Manually change the circuit breaker's state to 'closed', 'half-open', or 'open' using dedicated methods.

```python
# Closes the circuit
db_breaker.close()
```

```python
# Half-opens the circuit
db_breaker.half_open()
```

```python
# Opens the circuit
db_breaker.open()
```

--------------------------------

### Use Circuit Breaker with Decorator

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Apply the circuit breaker as a decorator to a function to automatically wrap its calls.

```python
@db_breaker
def update_customer(cust):
    # Do stuff here...
    pass
```

--------------------------------

### Exclude Exceptions from Failure Counting

Source: https://context7.com/danielfm/pybreaker/llms.txt

Use `exclude` to prevent specific exceptions or predicates from counting as failures. Exceptions can be dynamically added or removed.

```python
import pybreaker
from http import HTTPStatus

class UserNotFoundError(Exception):
    pass

class PaymentValidationError(Exception):
    pass

# Exclude by type (and all subclasses)
breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    exclude=[UserNotFoundError, PaymentValidationError],
)

# Exclude by predicate — only 4xx HTTP errors are business errors
http_breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    exclude=[lambda e: isinstance(e, IOError) and getattr(e, "status_code", 500) < 500],
)

# Mix types and callables
mixed_breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    exclude=[UserNotFoundError, lambda e: isinstance(e, ValueError) and "validation" in str(e)],
)

# Dynamic management
breaker.add_excluded_exception(LookupError)
breaker.add_excluded_exceptions(KeyError, IndexError)
breaker.remove_excluded_exception(LookupError)
print(breaker.excluded_exceptions)  # (UserNotFoundError, PaymentValidationError, KeyError, IndexError)
```

--------------------------------

### Use Circuit Breaker with Asynchronous Tornado Functions (Decorator)

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Circuit breakers can wrap asynchronous Tornado functions using a decorator with __pybreaker_call_async=True.

```python
from tornado import gen

@db_breaker(__pybreaker_call_async=True)
@gen.coroutine
def async_update(cust):
    # Do async stuff here...
    pass
```

--------------------------------

### CircuitBreaker as a Decorator

Source: https://context7.com/danielfm/pybreaker/llms.txt

Apply a CircuitBreaker instance as a decorator to protect functions. This is a common pattern for integrating with external services. functools.wraps is used to preserve the original function's metadata.

```python
import pybreaker

api_breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=60)

@api_breaker
def get_payment_status(payment_id: str) -> dict:
    """Fetch payment status from the payment gateway."""
    # ... real HTTP call here ...
    raise TimeoutError("Gateway timeout")

try:
    status = get_payment_status("pay_123")
except TimeoutError:
    print("Call failed, failure counted")
except pybreaker.CircuitBreakerError:
    print("Circuit is open — gateway is down, skipping call")

print(get_payment_status.__name__)  # 'get_payment_status'
print(get_payment_status.__doc__)   # 'Fetch payment status from the payment gateway.'
print(api_breaker.fail_counter)     # 1
```

--------------------------------

### Define Custom CircuitBreaker Listener

Source: https://github.com/danielfm/pybreaker/blob/main/README.rst

Subclass CircuitBreakerListener to define custom actions for circuit breaker events like state changes, failures, and successes.

```python
class DBListener(pybreaker.CircuitBreakerListener):
    """Listener used by circuit breakers that execute database operations."""

    def before_call(self, cb, func, *args, **kwargs):
        """Called before the circuit breaker `cb` calls `func`. """
        pass

    def state_change(self, cb, old_state, new_state):
        """Called when the circuit breaker `cb` state changes."""
        pass

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

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

--------------------------------

### Implement CircuitBreakerListener for Event Hooks

Source: https://context7.com/danielfm/pybreaker/llms.txt

Subclass `CircuitBreakerListener` to add custom logic for logging, metrics, or alerting. Listeners can be added or removed at runtime.

```python
import logging
import pybreaker

logger = logging.getLogger(__name__)

class ObservabilityListener(pybreaker.CircuitBreakerListener):
    def before_call(self, cb, func, *args, **kwargs):
        logger.debug("CB '%s' calling %s", cb.name, func.__name__)

    def success(self, cb):
        logger.info("CB '%s' call succeeded (fails=%d)", cb.name, cb.fail_counter)

    def failure(self, cb, exc):
        logger.warning("CB '%s' call FAILED [%s] (fails=%d/%d)",
                       cb.name, type(exc).__name__, cb.fail_counter, cb.fail_max)

    def state_change(self, cb, old_state, new_state):
        logger.error("CB '%s' state: %s -> %s",
                     cb.name,
                     old_state.name if old_state else "none",
                     new_state.name)

breaker = pybreaker.CircuitBreaker(
    fail_max=3,
    name="payments",
    listeners=[ObservabilityListener()],
)

# Add / remove listeners at runtime
extra = pybreaker.CircuitBreakerListener()
breaker.add_listener(extra)
breaker.remove_listener(extra)
print(len(breaker.listeners))  # 1
```

--------------------------------

### Excluding Exceptions

Source: https://context7.com/danielfm/pybreaker/llms.txt

Configure the circuit breaker to exclude specific exceptions from being counted as failures. This can be done by providing a list of exception types or a callable predicate.

```APIDOC
## Excluding Exceptions

By default every exception is counted as a system failure. Use `exclude` to whitelist business/domain exceptions (or exceptions matching a callable predicate) so they are not counted against the failure threshold.

### Exclusion Methods
- **exclude**: Accepts a list of exception types or callable predicates.
- **add_excluded_exception(exc)**: Adds a single exception type to the exclusion list.
- **add_excluded_exceptions(*exceptions)**: Adds multiple exception types to the exclusion list.
- **remove_excluded_exception(exc)**: Removes an exception type from the exclusion list.

### Usage
```python
import pybreaker
from http import HTTPStatus

class UserNotFoundError(Exception):
    pass

class PaymentValidationError(Exception):
    pass

# Exclude by type (and all subclasses)
breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    exclude=[UserNotFoundError, PaymentValidationError],
)

# Exclude by predicate — only 4xx HTTP errors are business errors
http_breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    exclude=[lambda e: isinstance(e, IOError) and getattr(e, "status_code", 500) < 500],
)

# Mix types and callables
mixed_breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    exclude=[UserNotFoundError, lambda e: isinstance(e, ValueError) and "validation" in str(e)],
)

# Dynamic management
breaker.add_excluded_exception(LookupError)
breaker.add_excluded_exceptions(KeyError, IndexError)
breaker.remove_excluded_exception(LookupError)
print(breaker.excluded_exceptions)  # (UserNotFoundError, PaymentValidationError, KeyError, IndexError)
```
```

--------------------------------

### Circuit Breaker for Generator Functions

Source: https://context7.com/danielfm/pybreaker/llms.txt

Guards generator functions by counting failures raised during iteration and successes upon normal completion. Failures mid-stream are counted.

```python
import pybreaker

breaker = pybreaker.CircuitBreaker(fail_max=3)

@breaker
def stream_records(fail_on: int):
    """Yield records, optionally failing mid-stream."""
    yield 1
    yield 2
    if fail_on == 3:
        raise IOError("stream interrupted")
    yield 3

# Successful generator — no failure counted
gen = stream_records(fail_on=0)
records = list(gen)
print(records)              # [1, 2, 3]
print(breaker.fail_counter) # 0

# Failing generator — failure counted after interruption
gen2 = stream_records(fail_on=3)
try:
    for _ in gen2:
        pass
except IOError:
    pass
print(breaker.fail_counter) # 1
```

--------------------------------

### Manually Control Circuit Breaker State

Source: https://context7.com/danielfm/pybreaker/llms.txt

Override the circuit state manually for maintenance or health checks. State changes are logged, and counters are reset upon closing.

```python
import pybreaker

breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=30)

# Force open during a planned maintenance window
breaker.open()
print(breaker.current_state)  # 'open'

# Put into half-open to probe recovery
breaker.half_open()
print(breaker.current_state)  # 'half-open'

# Force close when the system is confirmed healthy
breaker.close()
print(breaker.current_state)  # 'closed'
print(breaker.fail_counter)   # 0  (reset on close)

# Runtime reconfiguration
breaker.fail_max = 10
breaker.reset_timeout = 120
breaker.success_threshold = 3
print(breaker.fail_max)          # 10
print(breaker.reset_timeout)     # 120
print(breaker.success_threshold) # 3
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.