openfoodfacts_proxy.infrastructure.rate_limiter
[docs]
module
openfoodfacts_proxy.infrastructure.rate_limiter
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
69
70
71
72
73
74
75 | from collections import defaultdict, deque
from threading import Lock
import time
from openfoodfacts_proxy.core.settings_models import RateLimitSettings
from openfoodfacts_proxy.infrastructure.settings import settings
class SlidingWindowRateLimiter:
"""In-memory sliding window rate limiter tracking requests per IP and bucket."""
def __init__(self, rate_limit_settings: RateLimitSettings | None = None) -> None:
self._requests: dict[str, deque[float]] = defaultdict(deque)
self._lock = Lock()
self._rate_limit_settings = rate_limit_settings or settings.rate_limit
def _get_key(self, ip: str, bucket: str) -> str:
return f"{ip}:{bucket}"
def _get_limit(self, bucket: str) -> int:
if bucket == "search":
return self._rate_limit_settings.search_rate_limit
return self._rate_limit_settings.product_rate_limit
def _evict_expired(self, key: str, now: float) -> deque[float]:
"""Remove timestamps outside the current window."""
bucket = self._requests[key]
cutoff = now - self._rate_limit_settings.rate_limit_window_seconds
while bucket and bucket[0] <= cutoff:
bucket.popleft()
return bucket
def is_within_limit(self, ip: str, bucket: str) -> bool:
"""Check if the request is within rate limit. Records the request if within limit.
Returns True if the request is allowed (within limit), False if rate limit exceeded.
"""
key = self._get_key(ip, bucket)
limit = self._get_limit(bucket)
now = time.monotonic()
with self._lock:
entries = self._evict_expired(key, now)
if len(entries) >= limit:
return False
entries.append(now)
return True
def get_remaining(self, ip: str, bucket: str) -> int:
"""Get the number of remaining requests allowed in the current window."""
key = self._get_key(ip, bucket)
limit = self._get_limit(bucket)
now = time.monotonic()
with self._lock:
entries = self._evict_expired(key, now)
return max(0, limit - len(entries))
def cleanup(self) -> None:
"""Remove all expired entries to free memory."""
now = time.monotonic()
with self._lock:
keys_to_delete = []
for key in self._requests:
self._evict_expired(key, now)
if not self._requests[key]:
keys_to_delete.append(key)
for key in keys_to_delete:
del self._requests[key]
rate_limiter = SlidingWindowRateLimiter()
|