37 lines
1.2 KiB
Python
37 lines
1.2 KiB
Python
"""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
|