Implement circuit breaker open after failures demo in Python
A minimal CircuitBreaker class that calls a function and automatically 'opens' after a set number of consecutive failures, blocking further calls with a RuntimeError.
Python code
38 linesimport time
from datetime import datetime
class CircuitBreaker:
def __init__(self, threshold=3):
self.threshold = threshold
self.failure_count = 0
self.is_open = False
def call(self, func, *args, **kwargs):
if self.is_open:
raise RuntimeError("Circuit is OPEN")
try:
result = func(*args, **kwargs)
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
if self.failure_count >= self.threshold:
self.is_open = True
raise e
def flaky_service():
print(f" [{datetime.now().strftime('%H:%M:%S')}] flaky_service called")
raise ConnectionError("Service unavailable")
if __name__ == "__main__":
circuit = CircuitBreaker(threshold=3)
for i in range(5):
try:
circuit.call(flaky_service)
except (ConnectionError, RuntimeError) as e:
print(f" attempt {i+1}: {type(e).__name__}: {e}")
print(f"\nCircuit open: {circuit.is_open}")
print("Sample demonstrates circuit breakers stop calling failing services after threshold")
Output
[10:00:01] flaky_service called
attempt 1: ConnectionError: Service unavailable
[10:00:01] flaky_service called
attempt 2: ConnectionError: Service unavailable
[10:00:01] flaky_service called
attempt 3: ConnectionError: Service unavailable
attempt 4: RuntimeError: Circuit is OPEN
attempt 5: RuntimeError: Circuit is OPEN
Circuit open: True
Sample demonstrates circuit breakers stop calling failing services after threshold
How it works
The CircuitBreaker.call wrapper first checks the is_open flag; if true, it raises RuntimeError immediately, so the failing service is never invoked again. Otherwise it runs the function, and on any exception it increments failure_count. Once failure_count reaches the threshold (default 3), is_open flips to True permanently for this demo. Each successful call resets the counter, so a breaker recovers automatically when the service starts succeeding. This is the core cache-aside style pattern used in production reliability layers to prevent cascading failures.
Common mistakes
- Forgetting to reset the failure count on success, so the breaker stays open permanently
- Not including a timeout recovery mechanism — real breakers half-open after a cooldown period
- Raising the original exception without also tracking consecutive failures separately from a global counter
Variations
- Add a `cooldown` parameter to automatically transition back to half-open after a sleep interval
- Use `functools.wraps` and wrap the callable with a decorator to keep function metadata
Real-world use cases
- Wrapping HTTP calls to a third-party API so repeated timeouts stop hammering an unhealthy endpoint.
- Guarding database connection attempts during an outage, letting the app fail fast and preserve resources.
- Protecting a queue consumer when a downstream worker is down, avoiding wasted retries and log spam.
Sponsored
More from Errors & debugging
- Catch RecursionError and Fail Gracefully in Python easy
- Catch ValueError and print friendly message in Python easy
- Collect Multiple Validation Errors in Python Before Raising medium
- Handle ValueError and ZeroDivisionError in Python with try except easy
- How to Add a Correlation ID to Logging Records in Python medium
- How to Assert Preconditions with Descriptive Messages in Python easy
Keep learning
Related tutorials and quizzes for this topic.