patrickandClaude Sonnet 5.5
347f55cf7c
Includes atomic migrations, per-rule reminder semantics, first-run helper script and the note on the undecided UI chat idea. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
import time
|
|
from collections import deque
|
|
from typing import Callable, Hashable
|
|
|
|
WINDOW_SECONDS = 60
|
|
|
|
|
|
class SendLimiter:
|
|
def __init__(self, clock: Callable[[], float] = time.time, window: float = WINDOW_SECONDS):
|
|
self._clock = clock
|
|
self._window = window
|
|
self._stamps: dict[Hashable, deque[float]] = {}
|
|
self._overflow: dict[Hashable, int] = {}
|
|
self._last_notice: dict[Hashable, float] = {}
|
|
|
|
def acquire(self, key: Hashable, limit: int, force: bool = False) -> bool:
|
|
now = self._clock()
|
|
stamps = self._stamps.setdefault(key, deque())
|
|
while stamps and now - stamps[0] >= self._window:
|
|
stamps.popleft()
|
|
if not force and len(stamps) >= limit:
|
|
return False
|
|
stamps.append(now)
|
|
return True
|
|
|
|
def overflowed(self, key: Hashable, count: int = 1) -> None:
|
|
self._overflow[key] = self._overflow.get(key, 0) + count
|
|
|
|
def due_notices(self) -> list[tuple[Hashable, int]]:
|
|
now = self._clock()
|
|
due = []
|
|
for key, count in list(self._overflow.items()):
|
|
if count <= 0 or now - self._last_notice.get(key, float("-inf")) < self._window:
|
|
continue
|
|
self._last_notice[key] = now
|
|
self._overflow[key] = 0
|
|
due.append((key, count))
|
|
return due
|