Circuit breaker failure threshold count in Python

Track consecutive or time-windowed failures with a deque to open a circuit breaker and auto-recover to half-open after a cooldown.

Medium Python 3.9+ Aug 9, 2026 Reliability & rate limiting 16 views 0 copies

circuit-breaker resilience deque failure-handling time-window

Python code

39 lines

Python 3.9+

from collections import deque
from time import time, sleep


class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, recovery_time: float = 10.0):
        self.failure_threshold = failure_threshold
        self.recovery_time = recovery_time
        self.failures: deque[float] = deque()
        self.state = "closed"
        self.opened_at: float | None = None

    def record_failure(self) -> None:
        now = time()
        self.failures.append(now)
        while self.failures and now - self.failures[0] > self.recovery_time:
            self.failures.popleft()
        if len(self.failures) >= self.failure_threshold:
            self.state = "open"
            self.opened_at = now

    def record_success(self) -> None:
        self.failures.clear()
        self.state = "closed"
        self.opened_at = None

    def attempt(self) -> str:
        if self.state == "open" and time() - self.opened_at >= self.recovery_time:
            self.state = "half-open"
        return self.state


if __name__ == "__main__":
    breaker = CircuitBreaker(failure_threshold=3, recovery_time=2)
    for i in range(1, 6):
        breaker.record_failure()
        print(f"Failure {i}: state={breaker.attempt()}, failures={len(breaker.failures)}")
    breaker.record_success()
    print(f"After success: state={breaker.attempt()}, failures={len(breaker.failures)}")

Output

stdout

Failure 1: state=closed, failures=1
Failure 2: state=closed, failures=2
Failure 3: state=open, failures=3
Failure 4: state=open, failures=4
Failure 5: state=open, failures=5
After success: state=closed, failures=0

How it works

The deque stores failure timestamps in order. When a new failure arrives, old entries older than the recovery window are dropped from the left, keeping only recent failures. When the deque length reaches the threshold, the breaker flips to open. record_success clears all failures and resets to closed. The attempt method promotes open to half-open once the recovery time has elapsed, letting a limited test call through. This pattern gives a clean, testable state machine without external dependencies.

Common mistakes

Counting total failures instead of only those inside the recovery window.
Forgetting to slide the window when old failures expire.
Not resetting `opened_at` on success, so `attempt` may incorrectly jump to half-open.
Using a fixed list and scanning it, which is O(n) per failure instead of O(1) with deque.

Variations

Track total consecutive failures (no time window) using a simple integer counter.
Use `collections.Counter` with timestamps bucketed by seconds for a sliding-window count.

Real-world use cases

Protecting an API client from repeatedly calling a failing dependency and overloading it.
Failing fast in microservices when a downstream service errors more than N times in a minute.
Pausing retries to a non-responsive database connection after repeated etimedout errors.

Circuit breaker failure threshold count in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Reliability & rate limiting

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Reliability & rate limiting

Keep learning

Tutorials

Quizzes