"""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