added rate-limiting for specific requests

This commit is contained in:
Ebbe Baß
2026-09-09 13:54:06 +02:00
parent 9d52d869fb
commit e10efaa129
3 changed files with 66 additions and 2 deletions
+36
View File
@@ -0,0 +1,36 @@
"""A minimal cooldown gate to stop a single endpoint from being spammed.
Not a general-purpose rate limiter - no per-client/IP tracking, no token
bucket. Just one shared "don't run again until N seconds have passed since
the last call" gate, which is all a low-traffic single-instance endpoint
like /reload needs.
"""
import threading
import time
from typing import Optional
class Cooldown:
"""Gate that allows one call per ``interval_seconds``, shared by all callers."""
def __init__(self, interval_seconds: float):
self._interval = interval_seconds
self._lock = threading.Lock()
self._last_call: Optional[float] = None
def try_acquire(self) -> Optional[float]:
"""Attempt to pass the gate.
Returns None if allowed (and records this call as the new last
call). Returns the number of seconds still left to wait if the
gate is still cooling down.
"""
now = time.monotonic()
with self._lock:
if self._last_call is not None:
remaining = self._interval - (now - self._last_call)
if remaining > 0:
return remaining
self._last_call = now
return None