PyBreaker

Warnings

• breaking Python 3.8 support was dropped in PyBreaker v1.3.0, and Python 3.7 support was dropped in v1.2.0. Users on these older Python versions must upgrade to Python 3.9+ to use newer PyBreaker versions.
  Fix: Upgrade your Python environment to 3.9 or newer.
• deprecated PyBreaker v1.3.0 migrated from `datetime.datetime.utcnow()` to `datetime.datetime.now(UTC)` due to `utcnow()` being deprecated in Python 3.12. While `pybreaker` itself handles this, users with custom listeners or state management using `utcnow()` might encounter `DeprecationWarning`s or unexpected behavior with naive datetimes in newer Python versions.
  Fix: Ensure all datetime operations use timezone-aware objects, preferably `datetime.datetime.now(datetime.timezone.utc)` or `datetime.datetime.now(UTC)` for current UTC time.
• gotcha When using `CircuitRedisStorage`, do NOT initialize the `redis.StrictRedis` (or `redis.Redis`) connection with `decode_responses=True`. This will cause `AttributeError: 'str' object has no attribute 'decode'` in Python 3+ when `pybreaker` attempts to decode state values.
  Fix: Remove `decode_responses=True` from your Redis client initialization when passing it to `CircuitRedisStorage`.
• gotcha For distributed applications with multiple instances of your service, a `CircuitBreaker` instance must use shared state storage (e.g., `CircuitRedisStorage`) to ensure all instances observe the same circuit state. Without shared storage, each instance will manage its own circuit independently, defeating the purpose of a distributed circuit breaker.
  Fix: Use `CircuitRedisStorage` (or another shared storage implementation) and ensure a unique `namespace` is provided for each distinct circuit breaker if multiple are used with the same Redis instance. Example: `state_storage=pybreaker.CircuitRedisStorage(pybreaker.STATE_CLOSED, redis_client, namespace='my_unique_breaker_name')`.
• gotcha Incorrectly configuring `fail_max`, `reset_timeout`, or `success_threshold` can lead to circuits tripping too easily (false positives) or failing to open quickly enough, exacerbating cascading failures.
  Fix: Carefully tune circuit breaker parameters based on the expected reliability of the external service and your application's tolerance for failures. Monitor circuit state and logs to refine these thresholds. Implement meaningful fallbacks when the circuit is open.

Install

• pip install pybreaker Install stable version
• pip install pybreaker[redis] Install with Redis support

Imports

• CircuitBreaker

  from pybreaker import CircuitBreaker

  The CircuitBreaker class is the primary entry point for creating and managing circuit breakers.
• CircuitRedisStorage

  from pybreaker import CircuitRedisStorage

  Used for externalizing circuit breaker state to Redis, enabling distributed circuit breakers. Requires 'redis' package.
• CircuitBreakerError

  from pybreaker import CircuitBreakerError

  Exception raised when a circuit is open and prevents an operation from executing. Essential for handling fallback logic.

Quickstart

This quickstart demonstrates how to apply a circuit breaker to protect calls to an unreliable external service using `pybreaker.CircuitBreaker` as a decorator. It simulates service failures and shows how the circuit transitions between `CLOSED`, `OPEN`, and `HALF_OPEN` states, with appropriate error handling and a fallback mechanism.

import pybreaker
import requests

# Simulate a flaky external service
_service_up = True

def external_service_call():
    global _service_up
    if _service_up:
        print("Calling external service...")
        # Simulate a network error or service unavailability
        if requests.get("http://localhost:9999/health", timeout=0.1).status_code != 200:
            raise requests.exceptions.ConnectionError("Service not reachable")
        return "Service data"
    else:
        print("Service is intentionally down (simulated).")
        raise requests.exceptions.ConnectionError("Service is down")

# Create a circuit breaker instance
# Opens after 3 failures within 60 seconds
# Stays open for 10 seconds before trying again (half-open)
my_breaker = pybreaker.CircuitBreaker(
    fail_max=3,
    reset_timeout=10,
    exclude=[requests.exceptions.HTTPError] # Exclude HTTP errors that are not system failures
)

@my_breaker
def protected_call():
    return external_service_call()

print("--- Starting Circuit Breaker Demo ---")

for i in range(10):
    print(f"\nAttempt {i+1}:")
    try:
        result = protected_call()
        print(f"Success: {result}")
        _service_up = True # Reset service state on success to show circuit closing
    except pybreaker.CircuitBreakerError:
        print("Circuit OPEN! Falling back to cached data or default.")
        _service_up = False # Keep service down if breaker opened
    except requests.exceptions.ConnectionError as e:
        print(f"Connection error: {e}. Circuit state: {my_breaker.current_state}")
        # Simulate service coming back up after a few failures
        if i == 5: # Make service available after 5 attempts
            _service_up = True
    except Exception as e:
        print(f"Unexpected error: {e}")

# A real application would not typically control _service_up like this in quickstart,
# but it illustrates the breaker's behavior.