Python › Exception Handling

How would you implement a circuit breaker pattern using exception handling, to stop repeatedly calling a service that's consistently failing?

Advanced

Answer

Track consecutive failure counts; once a threshold is exceeded, the circuit 'opens' and immediately raises a CircuitOpenError WITHOUT even attempting the real call for a cooldown period, protecting the failing service from being hammered with more requests — after the cooldown, allow a single 'trial' call through to test if the service has recovered.

class CircuitBreaker:
    def __init__(self, threshold=3, cooldown=30):
        self.failures = 0
        self.threshold = threshold
        self.cooldown = cooldown
        self.opened_at = None
    def call(self, func, *args):
        if self.opened_at and time.time() - self.opened_at < self.cooldown:
            raise RuntimeError("Circuit is open")
        try:
            result = func(*args)
            self.failures = 0
            self.opened_at = None
            return result
        except Exception:
            self.failures += 1
            if self.failures >= self.threshold:
                self.opened_at = time.time()
            raise

Real-world example Preventing a struggling downstream service (like a flaky third-party API) from being overwhelmed with continued retry attempts during an outage.

Common follow-ups: How would you extend this circuit breaker to support a 'half-open' state that allows a limited number of trial requests through?

Concurrency (asyncio/threading/multiprocessing)