Circuit Breaker Pattern in Python: Closed, Open, and Half-Open States
Implement a circuit breaker with closed, open, and half-open states to prevent repeated calls to failing services and allow recovery after a timeout.
Python code
43 linesclass CircuitBreaker:
def __init__(self, failure_threshold=3, timeout_seconds=5):
self.failure_threshold = failure_threshold
self.timeout_seconds = timeout_seconds
self.state = "closed"
self.failure_count = 0
self.last_failure_time = None
def record_success(self):
if self.state == "half-open":
self.state = "closed"
print(f"Success in half-open -> state: {self.state}")
self.failure_count = 0
def record_failure(self):
if self.state == "closed":
self.failure_count += 1
if self.failure_count >= self.failure_threshold:
self.state = "open"
self.last_failure_time = __import__("time").time()
print(f"Threshold reached -> state: {self.state}")
elif self.state == "half-open":
self.state = "open"
self.last_failure_time = __import__("time").time()
print(f"Failure in half-open -> state: {self.state}")
def check_state(self):
if self.state == "open" and self.last_failure_time is not None:
import time
if time.time() - self.last_failure_time >= self.timeout_seconds:
self.state = "half-open"
print(f"Timeout elapsed -> state: {self.state}")
return self.state
if __name__ == "__main__":
cb = CircuitBreaker(failure_threshold=2, timeout_seconds=1)
print(f"Initial state: {cb.state}")
cb.record_failure()
cb.record_failure()
cb.check_state()
cb.record_success()
cb.check_state()
Output
Initial state: closed
Threshold reached -> state: open
Timeout elapsed -> state: half-open
Success in half-open -> state: closed
closed
How it works
The circuit breaker tracks consecutive failures when closed. Once the failure threshold is reached, it transitions to the open state and records the failure time. While open, all calls are short-circuited (not shown here) and after the timeout period, the check_state method moves it to half-open. In half-open, a single success resets the breaker to closed, while a failure trips it back to open. This pattern protects downstream services from overload and allows automatic recovery.
Common mistakes
- Not resetting failure count on success in half-open state
- Using `sleep` instead of checking elapsed time, causing blocking
- Assuming state transitions happen automatically without calling `check_state`
Variations
- Add a `call` method that wraps external requests and automatically records success or failure.
- Use a separate timer thread to transition from open to half-open asynchronously.
Real-world use cases
- Wrap HTTP calls to a third-party API to avoid hammering a failing service with retries.
- Guard database operations in a microservice to prevent connection pool exhaustion during an outage.
- Protect a messaging system consumer from repeatedly processing a poison message that causes failures.
Sponsored
More from System design patterns
- Build a BFF (Backend for Frontend) Mock Aggregator in Python medium
- Builder pattern for mocking complex objects in Python easy
- Create a Data Helper Class in Python easy
- Domain Driven Design Aggregate Root Example in Python medium
- Facade Pattern in Python with Mock Simplification medium
- How to Aggregate Mock API Routes by Method in Python easy
Keep learning
Related tutorials and quizzes for this topic.