35 lines
1001 B
Python
35 lines
1001 B
Python
import time
|
|
from collections import defaultdict
|
|
from typing import Callable
|
|
|
|
MAX_FAILURES = 5
|
|
WINDOW_SECONDS = 300.0
|
|
|
|
|
|
class FailureLimiter:
|
|
def __init__(self, max_failures: int = MAX_FAILURES, window: float = WINDOW_SECONDS,
|
|
clock: Callable[[], float] = time.monotonic):
|
|
self._max = max_failures
|
|
self._window = window
|
|
self._clock = clock
|
|
self._hits: dict[str, list[float]] = defaultdict(list)
|
|
|
|
def _prune(self, key: str) -> list[float]:
|
|
cutoff = self._clock() - self._window
|
|
hits = [t for t in self._hits[key] if t > cutoff]
|
|
if hits:
|
|
self._hits[key] = hits
|
|
else:
|
|
self._hits.pop(key, None)
|
|
return hits
|
|
|
|
def blocked(self, key: str) -> bool:
|
|
return len(self._prune(key)) >= self._max
|
|
|
|
def record(self, key: str) -> None:
|
|
self._prune(key)
|
|
self._hits[key].append(self._clock())
|
|
|
|
def reset(self, key: str) -> None:
|
|
self._hits.pop(key, None)
|