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.

Medium Python 3.9+ Aug 9, 2026 Errors & debugging 13 views 0 copies

Python code

38 lines
Python 3.9+
import 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

stdout
[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

  1. Add a `cooldown` parameter to automatically transition back to half-open after a sleep interval
  2. 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

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Errors & debugging

Related tutorials and quizzes for this topic.