MediumRate LimitingPython 3
Token Bucket Rate Limiter
Model per-key API quotas with deterministic refill math and caller-friendly retry-after decisions.
35m4 sample tests5 hidden tests
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 aDecision.- Each key has an independent bucket.
- Tokens refill continuously based on elapsed time.
- If a request is blocked,
retry_afteris the seconds until enough tokens are available. - When a request is allowed:
reason == "allowed"andretry_after == 0. - When a request is blocked:
reason == "rate_limited"andretry_after == (cost - tokens) / refill_rate_per_second(using the bucket's current tokens after refill). - Assume
cost <= capacityandrefill_rate_per_second > 0(callers must not pass larger costs or a zero rate; those inputs are out of scope).
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
7assert blocked.reason == "rate_limited"
8full = TokenBucketLimiter(3, 1, now=lambda: 0).allow("org", cost=3)
9assert (full.allowed, full.retry_after, full.reason) == (True, 0, "allowed")Constraints
- Use deterministic injected time.
- Use standard-library Python only.
Editor