Five rate-limiting algorithms, one clean interface, zero dependencies.
Rate limiting is the classic system-design interview question — "how would you build it?" This library answers it five ways, each thread-safe, per-key, and with an injectable clock so the time-based behaviour is deterministically testable.
from pyratelimit import TokenBucket
limiter = TokenBucket(rate=10, capacity=10) # 10 req/s, bursts up to 10
if limiter.allow("user:42"):
handle_request()| Algorithm | Idea | Bursts? | Memory | allow() |
|---|---|---|---|---|
| TokenBucket | tokens refill at a fixed rate, spend 1 per request | ✅ up to capacity | O(1)/key | O(1) |
| LeakyBucket | requests fill a bucket that leaks at a fixed rate | ❌ smooth | O(1)/key | O(1) |
| FixedWindowCounter | count requests per fixed time window | O(1)/key | O(1) | |
| SlidingWindowLog | exact log of timestamps in the window | ❌ exact | O(limit)/key | O(1) amortized |
| SlidingWindowCounter | weighted current+previous window (approx) | ❌ smooth | O(1)/key | O(1) |
Each implements the same interface:
limiter.allow(key="global", cost=1.0) -> bool # non-blocking check
limiter.try_acquire(key, cost) # raises RateLimitExceeded- Fixed window is O(1) but allows 2× bursts at the window boundary — so we also ship the sliding window log (exact, but O(limit) memory) and the sliding window counter (O(1) memory, smooths the boundary by weighting the previous window). Knowing that trade-off is the point of the question.
- Token vs leaky bucket: same math, mirrored — one allows bursts, one enforces a smooth rate. The code makes the duality obvious.
- Every limiter takes a
time_funcso tests don't sleep — they advance a fake clock. That's how you test time-based code.
from pyratelimit import SlidingWindowLog, rate_limited
api = SlidingWindowLog(limit=100, window=60) # 100 calls / minute
@rate_limited(api, key=lambda req: req.user_id)
def handler(req): ...pip install pyratelimit
python -m pytest # 100% deterministic, no sleeps
python examples/demo.py # see all five side by sideMIT