Spaces:
Sleeping
Sleeping
File size: 2,627 Bytes
ba2278a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | """A minimal per-key circuit breaker for outbound scraping backends.
When a backend (e.g. one search engine) is rate-limiting or blocking us,
retrying it on every single query just makes that worse - it's the
opposite of what you want when the goal is to stop looking suspicious to
that backend. This tracks consecutive failures per key and, once a
threshold is hit, short-circuits further calls to that key for a cooldown
period instead of attempting (and paying the timeout cost of) a call that
is very likely to fail anyway. After the cooldown it lets exactly one
trial call through (half-open); success closes the circuit again, failure
re-opens it for another full cooldown.
Not thread-safe (no locking) - fine given every caller in this codebase
only ever runs on a single asyncio event loop.
"""
import time
from dataclasses import dataclass
class CircuitOpenError(Exception):
"""Raised by `before_call` instead of letting a call through while a
key's circuit is open."""
def __init__(self, key: str, retry_after: float):
self.key = key
self.retry_after = retry_after
super().__init__(f"Circuit open for '{key}'; retry after {retry_after:.0f}s")
@dataclass
class _BreakerState:
failure_count: int = 0
opened_at: float = 0.0
class CircuitBreaker:
def __init__(self, failure_threshold: int = 3, cooldown_seconds: float = 60.0):
self.failure_threshold = failure_threshold
self.cooldown_seconds = cooldown_seconds
self._states: dict[str, _BreakerState] = {}
def _state(self, key: str) -> _BreakerState:
return self._states.setdefault(key, _BreakerState())
def before_call(self, key: str) -> None:
"""Raise CircuitOpenError if `key` is currently open and its
cooldown hasn't elapsed. Call this before attempting the call.
"""
state = self._state(key)
if state.failure_count < self.failure_threshold:
return
remaining = self.cooldown_seconds - (time.monotonic() - state.opened_at)
if remaining > 0:
raise CircuitOpenError(key, remaining)
# Cooldown elapsed: half-open - let this one call through. A
# renewed failure (record_failure) re-opens it for another cooldown;
# a success (record_success) clears it entirely.
def record_success(self, key: str) -> None:
self._states.pop(key, None)
def record_failure(self, key: str) -> None:
state = self._state(key)
state.failure_count += 1
if state.failure_count >= self.failure_threshold:
state.opened_at = time.monotonic()
|