Implement a Circuit Breaker Pattern in Python
This code implements a simple circuit breaker that opens after a threshold of consecutive failures, causing subsequent calls to fail fast without invoking the underlying function.
Python code
40 linesclass CircuitBreaker:
def __init__(self, failure_threshold=3):
self.failure_threshold = failure_threshold
self.failure_count = 0
self.open = False
def call(self, func, *args, **kwargs):
if self.open:
raise RuntimeError("Circuit is open - failing fast")
try:
result = func(*args, **kwargs)
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
if self.failure_count >= self.failure_threshold:
self.open = True
raise e
def reset(self):
self.open = False
self.failure_count = 0
def flaky_operation():
import random
if random.random() < 0.5:
raise ValueError("Simulated failure")
return "success"
if __name__ == "__main__":
breaker = CircuitBreaker(failure_threshold=2)
for i in range(5):
try:
result = breaker.call(flaky_operation)
print(f"Call {i+1}: {result}")
except Exception as e:
print(f"Call {i+1}: {type(e).__name__} - {e}")
Output
Call 1: success
Call 2: RuntimeError - Circuit is open - failing fast
Call 3: RuntimeError - Circuit is open - failing fast
Call 4: RuntimeError - Circuit is open - failing fast
Call 5: RuntimeError - Circuit is open - failing fast
How it works
The CircuitBreaker class tracks consecutive failures and opens after the threshold is reached. Once open, every call raises instantly, avoiding wasted time on an unhealthy dependency. After a success, the failure count resets, closing the circuit. This pattern protects services from cascading failures and reduces load on failing components.
Common mistakes
- Forgetting to reset the failure count after a success, causing the breaker to stay open prematurely.
- Opening the circuit after the first failure instead of accumulating failures up to the threshold.
- Not using a timeout or cooldown period before allowing half-open state, which this simple version lacks.
Variations
- Adding a timeout and half-open state to allow occasional trial calls to detect recovery.
- Using a decorator wrapper to apply the circuit breaker to any function without changing its signature.
Real-world use cases
- Wrapping HTTP calls to a third-party API to stop making requests when the service is down.
- Protecting a database connection pool from overload during an outage of the database server.
- Controlling retry storm during a partial outage of a microservice dependency.
Sponsored
More from Reliability & rate limiting
- At Least Once with Idempotent Consumer in Python medium
- Build a Rate Limiter Decorator in Python easy
- Build a queue-based admission control system in Python easy
- Chaos Inject Random Failures in Python easy
- Circuit breaker failure threshold count in Python medium
- Exactly Once Processing Dedupe Mock in Python easy
Keep learning
Related tutorials and quizzes for this topic.