circuit-breaker

name	Circuit Breaker
description	Resilience pattern preventing cascade failures by failing fast when a service is unavailable
category	software-development

Circuit Breaker

What I do

I provide a resilience pattern that prevents cascade failures in distributed systems. The circuit breaker monitors calls to external services and tracks failures. When failures exceed a threshold, the circuit "opens," immediately failing requests without calling the failing service. This allows the failing service time to recover while preventing resource exhaustion. After a cooldown period, the circuit allows test requests to determine if the service has recovered.

When to use me

Use circuit breakers when your application depends on external services that might fail or become slow. They're essential in microservice architectures, when calling third-party APIs, or when database connections might time out. Circuit breakers protect against cascade failures and help systems degrade gracefully. Don't use them for internal operations that are always fast or when you always want to attempt the call.

Core Concepts

• Closed State: Normal operation, requests pass through
• Open State: Failure threshold exceeded, requests fail fast
• Half-Open State: Testing if service has recovered
• Failure Threshold: Number or percentage of failures to trigger
• Timeout: Time before attempting recovery
• Failure Count: Tracking consecutive failures
• Success Count: Tracking successful calls in half-open state
• Fallback: Alternative behavior when circuit is open
• State Transitions: Rules for moving between states
• Metrics: Monitoring circuit state changes

Code Examples

Basic Circuit Breaker Implementation

from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
from typing import Callable, TypeVar, Generic
from uuid import uuid4
import time

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

T = TypeVar("T")

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5
    success_threshold: int = 3
    timeout_seconds: float = 60.0
    half_open_max_calls: int = 3

@dataclass
class CircuitBreakerMetrics:
    state: CircuitState
    failure_count: int = 0
    success_count: int = 0
    total_calls: int = 0
    last_failure_time: datetime | None = None
    last_state_change: datetime = datetime.utcnow()

class CircuitBreaker:
    def __init__(self, name: , config: CircuitBreakerConfig = ):
        ._name = name
        ._config = config  CircuitBreakerConfig()
        ._state = CircuitState.CLOSED
        ._failure_count = 
        ._success_count = 
        ._last_failure_time: datetime |  = 
        ._last_state_change = datetime.utcnow()
        ._half_open_calls = 
    

     () -> :
         ._name
    

     () -> CircuitState:
        ._check_state_transition()
         ._state
    

     () -> CircuitBreakerMetrics:
         CircuitBreakerMetrics(
            state=.state,
            failure_count=._failure_count,
            success_count=._success_count,
            total_calls=._failure_count + ._success_count,
            last_failure_time=._last_failure_time,
            last_state_change=._last_state_change
        )
    
     () -> T:
         .state == CircuitState.OPEN:
             CircuitOpenError(
                
            )
        
        :
            result = func(*args, **kwargs)
            ._on_success()
             result
        
         Exception  e:
            ._on_failure()
            
    
     () -> :
         ._state == CircuitState.HALF_OPEN:
            ._success_count += 
            ._half_open_calls += 
            
             ._success_count >= ._config.success_threshold:
                ._transition_to(CircuitState.CLOSED)
        
         ._state == CircuitState.CLOSED:
            ._failure_count = 
    
     () -> :
        ._failure_count += 
        ._last_failure_time = datetime.utcnow()
        
         ._state == CircuitState.HALF_OPEN:
            ._transition_to(CircuitState.OPEN)
        
         (
            ._state == CircuitState.CLOSED 
            ._failure_count >= ._config.failure_threshold
        ):
            ._transition_to(CircuitState.OPEN)
    
     () -> :
         ._state == CircuitState.OPEN:
            elapsed = datetime.utcnow() - ._last_state_change
             elapsed.total_seconds() >= ._config.timeout_seconds:
                ._transition_to(CircuitState.HALF_OPEN)
    
     () -> :
        old_state = ._state
        ._state = new_state
        ._last_state_change = datetime.utcnow()
        
         new_state == CircuitState.CLOSED:
            ._failure_count = 
            ._success_count = 
        
         new_state == CircuitState.HALF_OPEN:
            ._success_count = 
            ._half_open_calls = 
        
        ()

 ():
    

 ():
     ():
        .circuit_name = circuit_name
        ().__init__()

Decorator-Based Circuit Breaker

from functools import wraps
from typing import Callable, TypeVar, Generic

F = TypeVar("F", bound=Callable)

class CircuitBreakerDecorator:
    def __init__(self, circuit: CircuitBreaker):
        self._circuit = circuit
    
    def __call__(self, func: F) -> F:
        @wraps(func)
        def wrapper(*args, **kwargs):
            return self._circuit.call(func, *args, **kwargs)
        return wrapper

def circuit_breaker(
    name: str,
    failure_threshold: int = 5,
    timeout_seconds: float = 60
):
    circuit = CircuitBreaker(
        name=name,
        config=CircuitBreakerConfig(
            failure_threshold=failure_threshold,
            timeout_seconds=timeout_seconds
        )
    )
    decorator = CircuitBreakerDecorator(circuit)
    
    def actual_decorator(func: F) -> F:
        return decorator(func)
    
    return actual_decorator

@circuit_breaker(name="payment-service", failure_threshold=3)
def process_payment(order_id: str, amount: ) -> :
    ()
     amount < :
         ValueError()
     {: order_id, : , : amount}


 () -> :
    ()
     {: user_id, : }

 i  ():
    :
        result = process_payment(, )
        ()
     CircuitOpenError  e:
        ()
     Exception  e:
        ()

Circuit Breaker with Fallback

from abc import ABC, abstractmethod
from typing import Callable, Optional, Generic

class FallbackHandler(ABC):
    @abstractmethod
    def get_fallback(self, circuit_name: str, error: Exception) -> object:
        pass

class CircuitBreakerWithFallback(CircuitBreaker):
    def __init__(
        self,
        name: str,
        config: CircuitBreakerConfig = None,
        fallback_handler: FallbackHandler | None = None
    ):
        super().__init__(name, config)
        self._fallback_handler = fallback_handler
    
    def call(
        self,
        func: Callable[..., T],
        *args,
        fallback: Optional[Callable[..., T]] = None,
        **kwargs
    ) -> T:
        try:
            return super().call(func, *args, **kwargs)
        except CircuitOpenError:
            if fallback:
                return fallback()
            if self._fallback_handler:
                return self._fallback_handler.get_fallback(._name, CircuitOpenError())
            

 ():
     ():
        ._fallback_results = {}
    
     () -> :
        ._fallback_results[circuit_name] = result
    
     () -> :
         ._fallback_results.get(circuit_name, {: })

 :
     ():
        ._circuit = CircuitBreakerWithFallback(
            name=,
            fallback_handler=fallback_handler
        )
    
     () -> :
         ():
             ._actual_payment(order_id, amount)
        
         ():
             {: order_id, : , : }
        
         ._circuit.call(
            primary_payment,
            fallback=fallback_payment
        )
    
     () -> :
         amount > :
             ValueError()
         {: order_id, : , : amount}

Multi-Circuit Breaker Manager

from dataclasses import dataclass
from datetime import datetime
from typing import dict

@dataclass
class CircuitSummary:
    name: str
    state: CircuitState
    failure_count: int
    success_count: int
    last_state_change: datetime

class CircuitManager:
    def __init__(self):
        self._circuits: dict[str, CircuitBreaker] = {}
    
    def get_circuit(self, name: str, config: CircuitBreakerConfig = None) -> CircuitBreaker:
        if name not in self._circuits:
            self._circuits[name] = CircuitBreaker(name, config)
        return self._circuits[name]
    
    def remove_circuit(self, name: str) -> None:
        if name in self._circuits:
            del self._circuits[name]
    
    def get_all_states(self) -> list[CircuitSummary]:
        return [
            CircuitSummary(
                name=name,
                state=circuit.state,
                failure_count=circuit.metrics.failure_count,
                success_count=circuit.metrics.success_count,
                last_state_change=circuit.metrics.last_state_change
            )
             name, circuit  ._circuits.items()
        ]
    
     () -> :
         circuit  ._circuits.values():
            circuit._transition_to(CircuitState.CLOSED)

 :
     ():
        ._manager = manager
    
     () -> CircuitBreaker:
         ._manager.get_circuit(name)
    
     ():
         
    
     ():
        

circuit_manager = CircuitManager()
circuits = CircuitBreakerContext(circuit_manager)

circuits[].call(process_payment, , )
circuits[].call(get_user, )

Async Circuit Breaker

import asyncio
from typing import Callable, Awaitable

class AsyncCircuitBreaker:
    def __init__(self, name: str, config: CircuitBreakerConfig = None):
        self._name = name
        self._config = config or CircuitBreakerConfig()
        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._success_count = 0
        self._last_failure_time: datetime | None = None
        self._last_state_change = datetime.utcnow()
        self._lock = asyncio.Lock()
    
    @property
    def state(self) -> CircuitState:
        return self._state
    
    async def call(
        self,
        func: Callable[..., Awaitable[T]],
        *args,
        **kwargs
    ) -> T:
        async with self._lock:
            if self._state == CircuitState.OPEN:
                if self._should_attempt_reset():
                    self._state = CircuitState.HALF_OPEN
                else:
                     CircuitOpenError()
        
        :
            result =  func(*args, **kwargs)
             ._on_success()
             result
        
         Exception  e:
             ._on_failure()
            
    
      () -> :
        elapsed = datetime.utcnow() - ._last_state_change
         elapsed.total_seconds() >= ._config.timeout_seconds
    
      () -> :
          ._lock:
             ._state == CircuitState.HALF_OPEN:
                ._success_count += 
                 ._success_count >= ._config.success_threshold:
                    ._transition_to(CircuitState.CLOSED)
            
             ._state == CircuitState.CLOSED:
                ._failure_count = 
    
      () -> :
          ._lock:
            ._failure_count += 
            ._last_failure_time = datetime.utcnow()
            
             ._state == CircuitState.HALF_OPEN:
                ._transition_to(CircuitState.OPEN)
            
             ._failure_count >= ._config.failure_threshold:
                ._transition_to(CircuitState.OPEN)
    
     () -> :
        old_state = ._state
        ._state = new_state
        ._last_state_change = datetime.utcnow()
        
         new_state == CircuitState.CLOSED:
            ._failure_count = 
            ._success_count = 
        
        ()

  () -> :
     asyncio.sleep()
     {: order_id, : }

  ():
    breaker = AsyncCircuitBreaker()
    
     i  ():
        :
            result =  breaker.call(async_payment_service, )
            ()
         CircuitOpenError:
            ()
         Exception  e:
            ()

asyncio.run(main())

Metrics and Monitoring

from dataclasses import dataclass
from datetime import datetime
from collections import defaultdict
import threading

@dataclass
class CircuitEvent:
    circuit_name: str
    from_state: CircuitState
    to_state: CircuitState
    timestamp: datetime
    details: str | None = None

class CircuitMetricsCollector:
    def __init__(self):
        self._events: list[CircuitEvent] = []
        self._state_counts: dict[str, dict[CircuitState, int]] = defaultdict(lambda: defaultdict(int))
        self._lock = threading.Lock()
    
    def record_transition(
        self,
        circuit_name: str,
        from_state: CircuitState,
        to_state: CircuitState,
        details: str | None = None
    ) -> None:
        with self._lock:
            event = CircuitEvent(
                circuit_name=circuit_name,
                from_state=from_state,
                to_state=to_state,
                timestamp=datetime.utcnow(),
                details=details
            )
            self._events.append(event)
            self._state_counts[circuit_name][from_state] -= 1
            ._state_counts[circuit_name][to_state] += 
    
     () -> :
         ._lock:
             {
                : circuit_name,
                : ._circuits[circuit_name].state,
                : ._circuits[circuit_name].metrics.failure_count,
                : ._circuits[circuit_name].metrics.success_count,
                : [
                    e  e  ._events[-:]
                     e.circuit_name == circuit_name
                ]
            }
    
     () -> :
         {
            name: .get_circuit_health(name)
             name  ._circuits
        }

 ():
     ():
        ().__init__(name, config)
        ._metrics = metrics
    
     () -> :
        old_state = ._state
        ()._transition_to(new_state)
        
         ._metrics:
            ._metrics.record_transition(
                ._name,
                old_state,
                new_state
            )

Best Practices

1. Configure Thresholds: Set appropriate failure and timeout values
2. Multiple Circuits: Use separate circuits per dependency
3. Monitor Actively: Track circuit state changes and metrics
4. Graceful Degradation: Implement fallbacks for open circuits
5. Testing: Test all state transitions
6. Alerts: Notify when circuits open or close
7. Dashboard: Visualize circuit states
8. Default Fallbacks: Provide degraded functionality
9. Async Support: Handle async operations correctly
10. Reset Manually: Allow manual reset in emergencies
11. Gradual Recovery: Use half-open for safe recovery
12. Documentation: Document circuit configurations