Approach: implement a simple thread-safe circuit breaker with three states (CLOSED, OPEN, HALF-OPEN). It counts consecutive failures; when failures >= threshold it trips to OPEN for a cooldown period. After cooldown, it transitions to HALF-OPEN and allows a limited number of probe calls; on success it resets to CLOSED, on failure it goes back to OPEN.python
# circuit_breaker.py
import time
import threading
from enum import Enum, auto
from functools import wraps
class State(Enum):
CLOSED = auto()
OPEN = auto()
HALF_OPEN = auto()
class CircuitBreaker:
def __init__(self, max_failures=5, cooldown=10.0, max_half_open_probes=1):
self.max_failures = max_failures
self.cooldown = cooldown
self.max_half_open_probes = max_half_open_probes
self._state = State.CLOSED
self._fail_count = 0
self._opened_at = None
self._half_open_probes = 0
self._lock = threading.Lock()
def _now(self):
return time.time()
def _enter_open(self):
self._state = State.OPEN
self._opened_at = self._now()
self._half_open_probes = 0
def _enter_half_open(self):
self._state = State.HALF_OPEN
self._half_open_probes = 0
self._fail_count = 0
def _enter_closed(self):
self._state = State.CLOSED
self._fail_count = 0
self._opened_at = None
self._half_open_probes = 0
def _allow_request(self):
if self._state == State.CLOSED:
return True
if self._state == State.OPEN:
if (self._now() - self._opened_at) >= self.cooldown:
self._enter_half_open()
return True
return False
if self._state == State.HALF_OPEN:
if self._half_open_probes < self.max_half_open_probes:
self._half_open_probes += 1
return True
return False
def call(self, func, *args, **kwargs):
with self._lock:
allowed = self._allow_request()
if not allowed:
raise RuntimeError("CircuitBreakerOpen: request blocked")
# Execute without holding lock to avoid blocking others
try:
result = func(*args, **kwargs)
except Exception as exc:
with self._lock:
self._fail_count += 1
# if in HALF_OPEN and failure -> OPEN immediately
if self._state == State.HALF_OPEN:
self._enter_open()
elif self._fail_count >= self.max_failures:
self._enter_open()
raise
else:
with self._lock:
if self._state == State.HALF_OPEN:
# successful probe -> close
self._enter_closed()
else:
# reset failure counter on success
self._fail_count = 0
return result
# decorator helper
def decorate(self, func):
@wraps(func)
def wrapper(*args, **kwargs):
return self.call(func, *args, **kwargs)
return wrapper
python
import random, time
from circuit_breaker import CircuitBreaker
cb = CircuitBreaker(max_failures=3, cooldown=5, max_half_open_probes=1)
def flaky_downstream(x):
# simulate 60% failure
if random.random() < 0.6:
raise Exception("downstream error")
return f"ok:{x}"
protected = cb.decorate(flaky_downstream)
for i in range(20):
try:
print(i, protected(i))
except Exception as e:
print(i, "FAILED:", e)
time.sleep(0.5)
