Rate LimiterStep 4 of 5

Token Bucket

12345

Step 4 — Token Bucket

The token bucket is the workhorse of production rate limiting: it allows short bursts while capping the long-run rate. A bucket holds up to capacity tokens and refills at refill_rate tokens per second. Each request costs a token (or more); if the bucket has enough, spend them and allow — otherwise reject.

Implement:

  • allow(cost=1) — first refill based on time elapsed since the last call (tokens = min(capacity, tokens + elapsed * refill_rate)), then, if tokens >= cost, spend them and return True; else False

The bucket starts full, so it can absorb an initial burst of capacity requests, then settles to the steady refill rate.

Hints
Starter code
import time


class TokenBucket:
    def __init__(self, capacity, refill_rate, clock=time.time):
        self.capacity = capacity
        self.refill_rate = refill_rate    # tokens added per second
        self.clock = clock
        self.tokens = float(capacity)     # start full
        self._last = clock()

    def allow(self, cost=1):
        """Refill by elapsed time, then spend cost tokens if available."""
        # TODO:
        #   now = self.clock()
        #   elapsed = now - self._last
        #   self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        #   self._last = now
        #   if self.tokens >= cost: subtract cost and return True
        #   return False
        pass
Write it on a laptop

The editor and the test runner need a wide screen. Open this page on a desktop browser and your progress will be waiting.

PreviousNext step