LeetLLM
MediumRate LimitingPython 3

Token Bucket Rate Limiter

Model per-key API quotas with deterministic refill math and caller-friendly retry-after decisions.

35m1 sample tests1 hidden tests

Token Bucket Rate Limiter

Implement a per-key token bucket limiter.

Requirements

  • Define Decision(allowed, retry_after, reason).
  • Define TokenBucketLimiter(capacity, refill_rate_per_second, now).
  • allow(key, cost=1) returns a Decision.
  • Each key has an independent bucket.
  • Tokens refill continuously based on elapsed time.
  • If a request is blocked, retry_after is the seconds until enough tokens are available.

Example

python
1clock = {"t": 0.0} 2limiter = TokenBucketLimiter(capacity=2, refill_rate_per_second=1, now=lambda: clock["t"]) 3assert limiter.allow("org").allowed 4assert limiter.allow("org").allowed 5blocked = limiter.allow("org") 6assert not blocked.allowed and blocked.retry_after == 1

Constraints

  • Use deterministic injected time.
  • Use standard-library Python only.

Editor
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
Sample Tests
blocks when empty and returns retry_after
from solution import TokenBucketLimiter

clock = {"t": 0.0}
limiter = TokenBucketLimiter(capacity=2, refill_rate_per_second=1, now=lambda: clock["t"])
assert limiter.allow(("org1", "messages")).allowed
assert limiter.allow(("org1", "messages")).allowed
blocked = limiter.allow(("org1", "messages"))
assert not blocked.allowed
assert blocked.retry_after == 1
clock["t"] = 1.0
assert limiter.allow(("org1", "messages")).allowed
Results
Run sample tests or submit all tests.