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 limiter. Each caller owns a refillable balance, so one caller's burst must never spend another caller's capacity.

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 that starts at full capacity.
  • Tokens refill continuously based on elapsed time and never exceed capacity.
  • A blocked request doesn't consume tokens.
  • If now() moves backward, treat elapsed time as zero and keep the previous refill timestamp. Returning to that timestamp later must not create refill credit.
  • 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

The following clock trace starts with capacity 2 and a refill rate of 1 token per second:

TimeRequest costTokens before decisionDecisionTokens afterward
0.022allowed0
0.510.5blocked, retry after 0.50.5
1.011allowed0

The blocked middle request leaves its half-token intact. The runnable assertions below cover the full-bucket and empty-bucket boundaries:

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.
  • Preserve the last valid refill timestamp when the injected clock moves backward.

Editor