How to Implement a Circuit Breaker in Python

from dataclasses import dataclass
from datetime import datetime, timedelta
import time


@dataclass
class CircuitBreaker:
    failure_threshold: int = 3
    timeout_seconds: float = 5.0
    failures: int = 0
    state: str = "closed"
    last_failure: datetime = None

    def call(self, func):
        if self.state == "open":
            if datetime.now() - self.last_failure > timedelta(seconds=self.timeout_seconds):
                print("Circuit half-open: testing recovery")
                self.state = "half_open"
            else:
                raise Exception("Circuit is open - fast fail")

        try:
            result = func()
            if self.state == "half_open":
                print("Success in half-open: closing circuit")
                self.state = "closed"
                self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure = datetime.now()
            if self.failures >= self.failure_threshold:
                self.state = "open"
                print(f"Circuit open after {self.failures} failures")
            raise e

    def reset(self):
        self.failures = 0
        self.state = "closed"


if __name__ == "__main__":
    breaker = CircuitBreaker(failure_threshold=2, timeout_seconds=1.0)

    def flaky_service():
        if datetime.now().second % 2 == 0:
            raise ValueError("Transient error")
        return "ok"

    for attempt in range(6):
        try:
            print(f"Attempt {attempt}: {breaker.call(flaky_service)}")
        except Exception as e:
            print(f"Attempt {attempt}: raised {type(e).__name__}: {e}")
        time.sleep(0.3)