Token Bucket Rate Limiting in Python APIs

This article explores how to implement rate limiting in Python API endpoints using the token bucket algorithm. It covers the core mechanics of the algorithm, demonstrates a direct in-memory Python implementation utilizing lazy evaluation, and illustrates how to integrate this logic into web frameworks like FastAPI to protect endpoints from abuse and traffic spikes.

The Mechanics of the Token Bucket Algorithm

The token bucket algorithm controls the rate of traffic based on a simple metaphor:

  1. Bucket Capacity: A bucket holds a maximum number of tokens, defining the burst capacity.
  2. Refill Rate: Tokens are added to the bucket at a constant rate per second until the bucket reaches its maximum capacity.
  3. Consumption: Each incoming request requires one or more tokens to proceed. If sufficient tokens are present, the request consumes them and is processed. If the bucket is empty, the API denies the request, typically returning an HTTP 429 Too Many Requests status.

Instead of running background threads to increment tokens continuously, efficient implementations use lazy evaluation. The algorithm calculates how many tokens should have accumulated since the last request based on the elapsed time.

Pure Python Implementation

Below is a thread-safe, in-memory implementation of the token bucket using time.monotonic() for reliable time measurement.

import time
import threading

class TokenBucket:
    def __init__(self, capacity: int, refill_rate: float):
        self.capacity = capacity
        self.refill_rate = refill_rate  # Tokens added per second
        self.tokens = float(capacity)
        self.last_refill = time.monotonic()
        self.lock = threading.Lock()

    def consume(self, tokens: int = 1) -> bool:
        with self.lock:
            now = time.monotonic()
            elapsed = now - self.last_refill
            
            # Add tokens based on elapsed time
            self.tokens = min(
                self.capacity, 
                self.tokens + elapsed * self.refill_rate
            )
            self.last_refill = now

            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

Integrating with an API Framework (FastAPI)

In modern Python web frameworks like FastAPI, rate limiting can be applied per client IP address using a dependency injection pattern.

from fastapi import FastAPI, HTTPException, Request, status
from collections import defaultdict

app = FastAPI()

# Global store for buckets keyed by client IP
buckets = defaultdict(lambda: TokenBucket(capacity=10, refill_rate=2.0))

def rate_limiter(request: Request):
    client_ip = request.client.host
    bucket = buckets[client_ip]
    
    if not bucket.consume(1):
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Rate limit exceeded. Try again later."
        )

@app.get("/data", dependencies=[Depends(rate_limiter)])
def get_data():
    return {"message": "Request allowed"}

Scaling Beyond a Single Process

The in-memory approach operates within a single Python process. When running multi-worker ASGI/WSGI servers (such as Gunicorn or Uvicorn workers) or across multiple server instances, each process maintains an independent state, undermining the rate limit.

To scale horizontally: