rate_limit.py

import time
import threading
from collections import deque
from functools import wraps

class RateLimitExceeded(Exception):
    pass

class GlobalRateLimiter:
    def __init__(self, max_calls: int, window_seconds: int):
        self.max_calls = max_calls
        self.window = window_seconds
        self._lock = threading.Lock()
        self._events = deque()

    def allow(self) -> bool:
        now = time.time()
        with self._lock:
            while self._events and now - self._events[0] > self.window:
                self._events.popleft()
            if len(self._events) < self.max_calls:
                self._events.append(now)
                return True
            return False

    def enforce(self, func=None):
        def decorator(f):
            @wraps(f)
            def wrapper(*args, **kwargs):
                if not self.allow():
                    raise RateLimitExceeded(f"Rate limit exceeded ({self.max_calls} req/{self.window}s)")
                return f(*args, **kwargs)
            return wrapper
        return decorator if func is None else decorator(func)