Rate limiting

Three families. Token bucket for steady-state with bursts. Leaky bucket for smoothed output regardless of input. Sliding- window counter for “no more than N in any 60-second window”.

Token bucket

See Greedy for the canonical token-bucket implementation; it is the greedy admit-if-tokens-available algorithm.

Sliding window

A bounded deque of event timestamps, evicting anything older than the window.

from collections import deque
import time

class SlidingWindow:
    def __init__(self, max_n: int, window_s: float):
        self.max = max_n
        self.window = window_s
        self.events = deque()

    def admit(self) -> bool:
        now = time.monotonic()
        while self.events and now - self.events[0] > self.window:
            self.events.popleft()
        if len(self.events) >= self.max:
            return False
        self.events.append(now)
        return True

Leaky bucket

A queue with a fixed drain rate. Input bursts into the queue, output emerges at the drain rate. Use when smoothing matters more than admission decisions.

References

• Greedy for the token-bucket implementation.

• deque for the underlying ring buffer.