Tenacity Circuit Breaker Fallback

import random
import time

from tenacity import RetryCallState, retry, stop_after_attempt, wait_fixed


class CircuitBreaker:
    def __init__(self, failure_threshold: int = 3, cooldown: float = 5.0):
        self.failure_threshold = failure_threshold
        self.cooldown = cooldown
        self.failures = 0
        self.opened_at: float | None = None

    def allow(self) -> bool:
        if self.opened_at is None:
            return True
        if time.monotonic() - self.opened_at >= self.cooldown:
            self.opened_at = None  # 半开：放行试探请求
            self.failures = 0
            return True
        return False

    def record(self, success: bool) -> None:
        if success:
            self.failures = 0
            self.opened_at = None
        else:
            self.failures += 1
            if self.failures >= self.failure_threshold:
                self.opened_at = time.monotonic()


breaker = CircuitBreaker(failure_threshold=3, cooldown=5)
FALLBACK_RATES = {'usd': 1.0, 'eur': 0.92}


def after_attempts(retry_state: RetryCallState) -> None:
    breaker.record(retry_state.outcome.failed)


@retry(stop=stop_after_attempt(2), wait=wait_fixed(0.1),
       after=after_attempts, reraise=True)
def call_rate_api() -> dict:
    if random.random() < 0.6:
        raise ConnectionError('rates service flaky')
    return {'usd': 1.0, 'eur': 0.91}


def get_rates() -> tuple[dict, str]:
    if not breaker.allow():
        return FALLBACK_RATES, 'fallback-open'
    try:
        return call_rate_api(), 'live'
    except ConnectionError:
        return FALLBACK_RATES, 'fallback-error'


for _ in range(8):
    rates, source = get_rates()
    print('source:', source, 'usd:', rates['usd'])