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 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.
  • When a request is allowed: reason == "allowed" and retry_after == 0.
  • When a request is blocked: reason == "rate_limited" and retry_after == (cost - tokens) / refill_rate_per_second (using the bucket's current tokens after refill).
  • Assume cost <= capacity and refill_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