Python Circuit Breaker

2.1.3 · active · verified Thu Apr 09

A Python implementation of the Circuit Breaker pattern, designed to prevent applications from repeatedly trying to perform an operation that is likely to fail. It supports both synchronous and asynchronous functions and has been classified as a 'Critical Project' on PyPI. The current version is 2.1.3.

Warnings

breaking Version 2.0.0 completely dropped support for Python 2.x and early Python 3 versions. Applications running on Python 2.x or Python 3.x prior to 3.7 will break upon upgrade.
Fix: Upgrade your Python environment to 3.7 or newer before upgrading `circuitbreaker` to version 2.0.0 or later.
gotcha By default, the `@circuit` decorator (and `CircuitBreaker` class without `exceptions` specified) will trip on *any* `Exception`. This can be overly broad; it's often better to specify which exceptions should trip the circuit to avoid tripping on non-recoverable or expected errors.
Fix: Use the `exceptions` parameter in the `@circuit` decorator or `CircuitBreaker` constructor (e.g., `@circuit(exceptions=ConnectionError)`) to specify the types of exceptions that should cause the circuit to trip. Alternatively, use `exclude` for exceptions that should *not* trip the circuit.
gotcha Misconfiguring `fail_max`, `reset_timeout`, or `recovery_timeout` (for HALF_OPEN state) can lead to a circuit that never opens, opens too frequently, or stays open indefinitely. Understanding the state transitions is crucial.
Fix: Carefully review the documentation for `fail_max` (number of failures before opening), `reset_timeout` (time in seconds to wait before transitioning to HALF_OPEN), and `recovery_timeout` (time for HALF_OPEN to attempt recovery). Adjust these parameters based on the expected behavior and recovery time of your external service.

Install

pip install circuitbreaker Install stable version

Imports

circuit
```
from circuitbreaker import circuit
```
Commonly used as a decorator for functions (both sync and async) to apply circuit breaker logic.
CircuitBreaker
```
from circuitbreaker import CircuitBreaker
```
The class for creating custom circuit breaker instances or for more advanced control.
CircuitBreakerError
```
from circuitbreaker import CircuitBreakerError
```
The exception raised when the circuit is open and a protected function is called.

Quickstart

This example demonstrates how to use the `@circuit` decorator to protect a potentially unreliable synchronous function. It configures the circuit to trip after 3 failures (`fail_max=3`) and attempt to reset after 5 seconds (`reset_timeout=5`). It also shows how to exclude certain exceptions from tripping the circuit and handling both the underlying service errors and `CircuitBreakerError` when the circuit is open.

import time
from circuitbreaker import circuit

# Define a service call that might fail
def unreliable_service():
    # Simulate failure 75% of the time
    if time.time() % 4 < 3:
        raise ConnectionError("Service is currently unavailable!")
    print("Service call successful!")
    return "Data from service"

# Decorate the unreliable service call with a circuit breaker
# Trips after 3 failures, resets after 5 seconds
@circuit(fail_max=3, reset_timeout=5, exclude=[ValueError])
def get_data():
    return unreliable_service()

print("--- Circuit Breaker Quickstart ---")
for i in range(10):
    print(f"Attempt {i+1}:")
    try:
        result = get_data()
        print(f"  Result: {result}")
    except ConnectionError as e: # Handle the underlying service error
        print(f"  Caught service error: {e}")
    except Exception as e: # Catch CircuitBreakerError or other unexpected errors
        print(f"  Caught general error from circuit: {e}")
    time.sleep(1)

view raw JSON →