generated from kgod/ai-review-template
38 lines
1.1 KiB
Python
38 lines
1.1 KiB
Python
"""In-process sliding-window rate limiting for single-process deployments."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections import deque
|
|
from typing import Callable
|
|
|
|
|
|
class SlidingWindowRateLimiter:
|
|
"""Allow at most ``limit`` requests per key during a sliding time window."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
limit: int,
|
|
window_seconds: float,
|
|
clock: Callable[[], float] = time.monotonic,
|
|
) -> None:
|
|
if limit < 1:
|
|
raise ValueError("limit must be positive")
|
|
if window_seconds <= 0:
|
|
raise ValueError("window_seconds must be positive")
|
|
self.limit = limit
|
|
self.window_seconds = window_seconds
|
|
self.clock = clock
|
|
self._hits: dict[str, deque[float]] = {}
|
|
|
|
def allow(self, key: str) -> bool:
|
|
now = self.clock()
|
|
hits = self._hits.setdefault(key, deque())
|
|
while hits and now - hits[0] >= self.window_seconds:
|
|
hits.popleft()
|
|
if len(hits) >= self.limit:
|
|
return False
|
|
hits.append(now)
|
|
return True
|