Pattern 8: Circuit Breaker

The Problem: Cascading Failures

Scenario: You call a downstream service (API, DB). It's having issues (500 errors, timeouts).

Naive approach: Retry every request. While the service recovers, you're hammering it with more requests. Bad!

Circuit breaker: After N failures, stop calling for a bit. Let the service recover. Then try again.

States: Closed → Open → Half-Open → Closed

stateDiagram-v2 [*] --> Closed Closed --> Open: Too many failures
(e.g., 5+ errors) Open --> HalfOpen: Timeout elapsed
(e.g., after 30s) HalfOpen --> Closed: Test request succeeds
Service recovered! HalfOpen --> Open: Test request fails
Still broken note right of Closed ✅ Normal operation Requests flow through end note note right of Open ⛔ Service failing Fast-fail: reject immediately Don't hammer the service end note note right of HalfOpen 🔍 Testing recovery Allow ONE request through Check if service is back end note

Key benefit: Protect yourself AND the downstream service. Fast failure instead of timeout cascades.

Timeline: Circuit Breaker in Action

timeline title Circuit Breaker State Changes Over Time section t=0-20s: CLOSED (requests passing) Normal operation: Requests succeed section t=20-22s: Failures detected Request fails: 500 error Request fails: timeout Request fails: 500 error section t=22s: OPEN (fast-fail) Circuit opens: reject requests instantly Service stops getting hammered section t=22-52s: Waiting for recovery 30 seconds pass: service recovers section t=52s: HALF-OPEN (test request) Allow 1 test request: check if service back Test succeeds: ✅ Service recovered! section t=52+: Back to CLOSED Resume normal operation Requests flowing normally again

Implementation

import asyncio
import httpx
from enum import Enum
from dataclasses import dataclass
import time

class CircuitState(Enum):
    CLOSED = "closed"       # Normal operation
    OPEN = "open"           # Failing, reject requests
    HALF_OPEN = "half_open" # Testing if recovered

@dataclass
class CircuitBreaker:
    failure_threshold: int = 5       # failures before opening
    recovery_timeout: int = 30       # seconds before half-open

    def __init__(self):
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None

    def is_available(self) -> bool:
        """Check if circuit allows requests."""
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            # Check if timeout elapsed
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                return True  # Allow test request
            return False
        if self.state == CircuitState.HALF_OPEN:
            return True  # Allow test request

    def record_success(self):
        """Record successful call."""
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED

    def record_failure(self):
        """Record failed call."""
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

async def call_with_circuit_breaker(cb: CircuitBreaker, client: httpx.AsyncClient, url: str):
    if not cb.is_available():
        raise Exception("Circuit breaker is OPEN, service unavailable")

    try:
        response = await client.get(url, timeout=5)
        response.raise_for_status()
        cb.record_success()
        return response.json()
    except Exception as e:
        cb.record_failure()
        raise

async def main():
    cb = CircuitBreaker(failure_threshold=3, recovery_timeout=10)
    async with httpx.AsyncClient() as client:
        for i in range(100):
            try:
                result = await call_with_circuit_breaker(cb, client, "https://api.example.com/data")
                print(f"Success: {result}")
            except Exception as e:
                print(f"Failed: {e} (circuit state: {cb.state.value})")
            await asyncio.sleep(1)

asyncio.run(main())

Real-World: With Fallback

async def call_with_fallback(cb, client, url, fallback_fn):
    if not cb.is_available():
        # Circuit is OPEN, use fallback (cache, default value)
        return await fallback_fn()

    try:
        response = await client.get(url)
        cb.record_success()
        return response.json()
    except Exception as e:
        cb.record_failure()
        return await fallback_fn()  # Fallback on failure

# Usage in FastAPI
@app.get("/user/{user_id}")
async def get_user(user_id: int):
    return await call_with_fallback(
        cb,
        client,
        f"https://user-service.example.com/users/{user_id}",
        fallback_fn=lambda: {"id": user_id, "cached": True}  # Return cached/default
    )