python 61 lines · 7 steps

Building a circuit breaker in Python

A thread-safe decorator that stops calling a failing service until it has time to recover.

Explained by highlit
Advanced circuit-breaker resilience decorators concurrency state-machine
1import time
2import threading
3from enum import Enum
4from functools import wraps
5 
6 
7class State(Enum):
8 CLOSED = "closed"
9 OPEN = "open"
10 HALF_OPEN = "half_open"
11 
12 
13class CircuitOpenError(Exception):
14 pass
15 
16 
17class CircuitBreaker:
18 def __init__(self, failure_threshold=5, reset_timeout=30.0, expected=Exception):
19 self.failure_threshold = failure_threshold
20 self.reset_timeout = reset_timeout
21 self.expected = expected
22 self._state = State.CLOSED
23 self._failures = 0
24 self._opened_at = 0.0
25 self._lock = threading.Lock()
26 
27 def _allow_request(self):
28 with self._lock:
29 if self._state is State.OPEN:
30 if time.monotonic() - self._opened_at >= self.reset_timeout:
31 self._state = State.HALF_OPEN
32 return True
33 return False
34 return True
35 
36 def _on_success(self):
37 with self._lock:
38 self._failures = 0
39 self._state = State.CLOSED
40 
41 def _on_failure(self):
42 with self._lock:
43 self._failures += 1
44 if self._state is State.HALF_OPEN or self._failures >= self.failure_threshold:
45 self._state = State.OPEN
46 self._opened_at = time.monotonic()
47 
48 def __call__(self, func):
49 @wraps(func)
50 def wrapper(*args, **kwargs):
51 if not self._allow_request():
52 raise CircuitOpenError(f"{func.__name__} circuit is open")
53 try:
54 result = func(*args, **kwargs)
55 except self.expected:
56 self._on_failure()
57 raise
58 self._on_success()
59 return result
60 
61 return wrapper
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A circuit breaker trades a few fast failures for protecting a struggling downstream service from being hammered.
  2. 2The half-open state acts as a single probe that decides whether the circuit fully closes again or trips back open.
  3. 3Guarding shared counters and state transitions with a lock keeps the breaker correct under concurrent calls.

Related explainers

python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
go 
func (w *Watcher) resetDebounce(d time.Duration) {
	if !w.timer.Stop() {
		select {
		case <-w.timer.C:

Debouncing a stream of events in Go

debounce timers channels
Advanced 7 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 steps
python 
import secrets
 
from django.contrib.auth import authenticate, login
from django.core.cache import cache

Two-factor login with OTP in Django

two-factor-auth one-time-passwords caching
Intermediate 9 steps
rust 
use std::collections::VecDeque;
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
 

Building a counting semaphore in Rust

concurrency synchronization condition-variable
Advanced 9 steps

Share this explainer

Here's the card — post it anywhere.

Building a circuit breaker in Python — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code