Circuit Breaker in Python — Implementing Resilience Patterns with PyCircuitBreaker

In this tutorial, you will learn about Circuit Breaker in Python. We cover key concepts, practical examples, and best practices to help you master this topic.

Python circuit breaker implementation using pybreaker and custom classes provides decorator-based and context manager integration with popular libraries like requests and aiohttp, configurable thresholds, and event-driven monitoring for resilient Python applications.

flowchart TD
    P[Python App] -->|@circuit_breaker| CB[PyBreaker
Circuit Breaker]
    CB -->|Closed| Req[requests/aiohttp]
    CB -->|Open| Fallback[Fallback Handler]
    CB -->|Events| EventBus[(Event Bus)]
    EventBus --> Monitor[Monitor/Logger]
    Req -->|Success| OK[Return]
    Req -->|Exception| Fail[Count Failure]

What You'll Learn

• PyBreaker library usage
• Custom circuit breaker class
• Decorator-based integration
• Async circuit breaker support
• Event-driven monitoring

Why It Matters

Python applications in microservices need circuit breakers for resilient HTTP calls, database queries, and external API integrations. Python's async ecosystem adds complexity: circuit breakers must support asyncio, aiohttp, and event-driven patterns.

Real-World Use

DodaTech's Python backend uses pybreaker for all external HTTP calls. Circuit breakers protect the recommendations API, search service, and content delivery endpoints. When the recommendations service fails, pybreaker opens within 10 requests and serves cached recommendations.

PyBreaker Basic Usage

import pybreaker
import requests
import time

breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=30,
    name="search-api"
)

@breaker
def search_products(query):
    response = requests.get(
        f"https://api.example.com/search?q={query}",
        timeout=3
    )
    response.raise_for_status()
    return response.json()

for i in range(8):
    try:
        result = search_products("laptop")
        print(f"Search {i+1}: {len(result)} results")
    except Exception as e:
        print(f"Search {i+1}: {type(e).__name__}: {e}")
    time.sleep(0.1)

Expected output:

Search 1: 15 results
Search 2: 15 results
Search 3: 15 results
Search 4: 15 results
Search 5: 15 results
Search 6: pybreaker.CircuitBreakerError: CircuitBreaker 'search-api' is open
Search 7: pybreaker.CircuitBreakerError: CircuitBreaker 'search-api' is open
Search 8: pybreaker.CircuitBreakerError: CircuitBreaker 'search-api' is open

Custom Python Circuit Breaker

import time
import threading
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    def __init__(self, name, fail_max=5, reset_timeout=30, expected_exceptions=(Exception,)):
        self.name = name
        self.fail_max = fail_max
        self.reset_timeout = reset_timeout
        self.expected_exceptions = expected_exceptions
        self.state = CircuitState.CLOSED
        self.failures = 0
        self.last_failure = 0
        self.lock = threading.Lock()

    def call(self, fn, fallback=None, *args, **kwargs):
        with self.lock:
            if self.state == CircuitState.OPEN:
                if time.time() - self.last_failure > self.reset_timeout:
                    self.state = CircuitState.HALF_OPEN
                    print(f"[{self.name}] Half-open probe")
                else:
                    return self._handle_fallback(fallback)

        try:
            result = fn(*args, **kwargs)
            with self.lock:
                self.failures = 0
                if self.state == CircuitState.HALF_OPEN:
                    self.state = CircuitState.CLOSED
                    print(f"[{self.name}] Closed (recovered)")
            return result
        except self.expected_exceptions as e:
            with self.lock:
                self.failures += 1
                self.last_failure = time.time()
                if self.failures >= self.fail_max:
                    self.state = CircuitState.OPEN
                    print(f"[{self.name}] Open ({self.failures} failures)")
            return self._handle_fallback(fallback)

    def _handle_fallback(self, fallback):
        if fallback:
            return fallback()
        raise CircuitBreakerOpenError(self.name)

class CircuitBreakerOpenError(Exception):
    def __init__(self, name):
        super().__init__(f"Circuit breaker '{name}' is open")

cb = CircuitBreaker("payment-api", fail_max=3, reset_timeout=10)

def call_payment():
    import random
    if random.random() < 0.6:
        raise ConnectionError("Payment service unavailable")
    return {"status": "paid"}

for i in range(6):
    try:
        result = cb.call(call_payment)
        print(f"Call {i+1}: {result}")
    except CircuitBreakerOpenError as e:
        print(f"Call {i+1}: {e}")
    time.sleep(0.1)

Expected output:

[payment-api] Half-open probe
Call 1: {'status': 'paid'}
[payment-api] Open (3 failures)
Call 2: Circuit breaker 'payment-api' is open
...

Async Circuit Breaker

import asyncio
import pybreaker
import time

class AsyncCircuitBreaker(pybreaker.CircuitBreaker):
    async def call_async(self, fn, *args, **kwargs):
        if self.current_state == pybreaker.STATE_OPEN:
            if time.time() - self._last_failure > self.reset_timeout:
                self._state = pybreaker.STATE_HALF_OPEN
            else:
                raise pybreaker.CircuitBreakerError(self.name)

        try:
            result = await fn(*args, **kwargs)
            self._success()
            return result
        except Exception as e:
            self._failure()
            raise

breaker = AsyncCircuitBreaker(fail_max=3, reset_timeout=10)

async def fetch_data(url):
    await asyncio.sleep(0.1)
    raise ConnectionError(f"Failed to fetch {url}")

async def call_with_breaker():
    for i in range(5):
        try:
            result = await breaker.call_async(
                fetch_data, f"https://api.example.com/data/{i}"
            )
            print(f"Call {i+1}: {result}")
        except pybreaker.CircuitBreakerError as e:
            print(f"Call {i+1}: Circuit open, blocked")
        except Exception as e:
            print(f"Call {i+1}: {e}")

asyncio.run(call_with_breaker())

Expected output:

Call 1: Failed to fetch https://api.example.com/data/0
Call 2: Failed to fetch https://api.example.com/data/1
Call 3: Failed to fetch https://api.example.com/data/2
Call 4: Circuit open, blocked
Call 5: Circuit open, blocked

Common Mistakes

• Not using expected_exceptions -- catching all exceptions opens the circuit on 4xx client errors. Configure expected_exceptions to catch only server errors (ConnectionError, Timeout, 5xx responses).
• Circuit breaker per function instead of per service -- creating a circuit breaker for each API endpoint means one slow endpoint opens only its own breaker. But it also means many circuit breaker instances to manage. Create one per downstream service.
• Forgetting async support -- pybreaker's default @breaker decorator does not support async functions. Use async-aware circuit breakers or manually check state in async functions.
• No fallback for Python circuit breakers -- pybreaker raises CircuitBreakerError when open. Always provide fallbacks via except handlers or pybreaker's call_with_fallback method.
• Global circuit breaker state -- pybreaker state is in-memory per process. Multiple worker processes each have their own state. Use Redis-backed circuit breakers for distributed state if needed.

Practice Questions

1. How does pybreaker's decorator syntax work?
2. Why should you specify expected_exceptions in Python circuit breakers?
3. How do you implement async-aware circuit breakers?
4. What is the difference between per-function and per-service circuit breakers?
5. How do you handle fallbacks with pybreaker?

Challenge

Build a Python circuit breaker framework: (1) Redis-backed distributed circuit breaker state for multi-worker deployments, (2) async support with asyncio and aiohttp integration, (3) decorator and context manager interfaces, (4) configurable failure thresholds per exception type (e.g., timeout=3 failures, 5xx=5 failures), (5) Prometheus metrics export for circuit breaker state and call counts, (6) fallback chain support (try primary, then cache, then default response).

FAQ

Mini Project

Build a Python resilience library with: (1) circuit breaker (pybreaker-compatible API with Redis backend), (2) retry with exponential backoff and jitter, (3) bulkhead with Semaphore and thread pool modes, (4) rate limiter with token bucket, (5) all primitives support both sync and async (asyncio) modes, (6) Prometheus metrics for all primitives: state, call count, latency histogram, (7) FastAPI middleware that applies circuit breakers per downstream service, (8) CLI tool for viewing and resetting circuit breaker state.

What's Next

Continue with Go Implementation to learn Go circuit breaker patterns. Then explore Async Patterns for async-compatible circuit breakers.