Install & Compatibility
Where this runs
tested against v1.4.1 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.910 runs
installs and imports cleanly · install 0.0s · import 0.039s · 17.9MB
glibcpy 3.10–3.910 runs
installs and imports cleanly · install 1.6s · import 0.034s · 18MB
16MB installed
● package 16MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
CircuitBreaker
✓ from pybreaker import CircuitBreaker
✗ import pybreaker; breaker = pybreaker.CircuitBreaker() without explicitly importing the class
The CircuitBreaker class is the primary entry point for creating and managing circuit breakers.
CircuitRedisStorage
✓ from pybreaker import CircuitRedisStorage
✗ from pybreaker.storage import CircuitRedisStorage (older paths)
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.
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.
Debug
Known issues
breakingPython 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.fixUpgrade your Python environment to 3.9 or newer.
affects: >=1.2.0 (for Python 3.7), >=1.3.0 (for Python 3.8)
deprecatedPyBreaker 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.fixEnsure all datetime operations use timezone-aware objects, preferably `datetime.datetime.now(datetime.timezone.utc)` or `datetime.datetime.now(UTC)` for current UTC time.
affects: >=1.3.0
gotchaWhen 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.fixRemove `decode_responses=True` from your Redis client initialization when passing it to `CircuitRedisStorage`.
affects: All versions using `CircuitRedisStorage`
gotchaFor 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.fixUse `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')`.
affects: All versions in distributed environments
gotchaIncorrectly 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.fixCarefully 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.
affects: All versions
gotchaApplications or examples using PyBreaker might encounter `ModuleNotFoundError` if external dependencies required for their specific functionality (e.g., `requests` for HTTP interactions, `redis` for `CircuitRedisStorage`) are not installed. Ensure all necessary third-party libraries are explicitly installed alongside PyBreaker.fixInstall all required external dependencies for your application or specific PyBreaker components. For example, `pip install requests` for HTTP client functionality, or `pip install pybreaker[redis]` for Redis storage.
affects: All versions
gotchaThe test script attempts to `import requests`, but `requests` is not a direct dependency of `pybreaker` and was not found in the environment. This `ModuleNotFoundError` indicates that `requests` needs to be explicitly installed if your application or test environment depends on it alongside `pybreaker`.fixEnsure `requests` is installed in your environment if your application or test script uses it (e.g., `pip install requests`).
affects: All versions
Errors
Common errors & fixes
ModuleNotFoundError: No module named 'pybreaker'
The 'pybreaker' package is not installed in your Python environment or the Python interpreter cannot locate it in its search path.
fixInstall the package using pip: `pip install pybreaker`
pybreaker.CircuitBreakerError
This error is raised by `pybreaker` when the circuit is in the 'OPEN' state, meaning the guarded function has consistently failed and the circuit breaker is preventing further calls to protect the system.
fixWrap calls to the protected function in a try-except block to catch `pybreaker.CircuitBreakerError` and implement appropriate fallback logic or inform the user that the service is unavailable. You can also configure a `fallback_function` when initializing the `CircuitBreaker`.
AttributeError: 'str' object has no attribute 'decode'
This error occurs when using `pybreaker.CircuitRedisStorage` if the `redis.StrictRedis` connection is initialized with `decode_responses=True`, which causes Redis to return string objects that `pybreaker` then attempts to decode again.
fixEnsure that the `redis.StrictRedis` client passed to `pybreaker.CircuitRedisStorage` is *not* initialized with `decode_responses=True`.
```python
import pybreaker
import redis
# INCORRECT: redis.StrictRedis(decode_responses=True)
# CORRECT:
redis_client = redis.StrictRedis()
db_breaker = pybreaker.CircuitBreaker(
fail_max=5,
reset_timeout=60,
state_storage=pybreaker.CircuitRedisStorage(pybreaker.STATE_CLOSED, redis_client)
)
``` redis.exceptions.ConnectionError: Error while reading from socket: ('Connection closed by server.',)
When using `pybreaker.CircuitRedisStorage`, this indicates a general connection issue with the Redis server, such as the server being down, unreachable due to network problems, or an incorrect Redis configuration.
fixVerify your Redis server is running and accessible from the application. Check network connectivity, firewall rules, and the Redis host/port configuration. You may also need to configure connection timeouts or implement retry logic at a lower level if Redis is intermittently unavailable.
Upgrade
Version history
1.4.1latest on PyPI · released Sep 21, 2025
Audit
Dependencies
pythonrequiredRequires Python 3.9 or newer.
redisoptionalOptional dependency for `CircuitRedisStorage` to enable shared state across multiple application instances.