Building a Token Bucket Rate Limiter in Python: A Step-by-Step Guide

A few months ago, one of our internal API endpoints started getting hammered by a misconfigured client. It wasn't an attack — just a loop bug that fired requests every 200ms instead of every 2 seconds. By the time we noticed, the endpoint was serving 429s to everyone else and our database connection pool was exhausted.

The fix wasn't a bigger server. It was a rate limiter.

In this post I'll show you how to build a token bucket rate limiter from scratch in Python — no third-party dependencies, about 40 lines of code, and easy to drop into FastAPI, Flask, or any WSGI app. Even if you end up using a library, understanding the algorithm will help you configure it correctly.

Why a token bucket and not a fixed window?

The naive approach is a fixed window: allow N requests per minute, reset the counter when the minute ends. It's simple, but it has a well-known flaw — the boundary problem. If a client sends 100 requests at 0:59 and another 100 at 1:01, they've sent 200 requests in 2 seconds, even though you allowed only 100 per minute.

A token bucket smooths this out. It allows bursts up to a defined capacity, while enforcing a steady long-term rate. That's exactly how most production APIs (and cloud load balancers) behave: you get a burst budget, but the average rate stays capped.

The algorithm in plain English

Think of a bucket that holds tokens:

• The bucket has a capacity (the maximum number of tokens it can hold).
• Tokens are added at a fixed rate — say 10 tokens per second.
• Each request removes one token (or more, if a request is expensive).
• If the bucket is empty, the request is rejected with 429.

Because the bucket starts full, a client can fire off capacity requests instantly (the burst), then settles into the steady rate after that.

The implementation

Here's a thread-safe token bucket in pure Python:

import threading
import time


class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate          # tokens added per second
        self.capacity = capacity  # maximum burst size
        self.tokens = float(capacity)
        self.updated_at = time.monotonic()
        self._lock = threading.Lock()

    def consume(self, tokens: int = 1) -> bool:
        with self._lock:
            now = time.monotonic()
            elapsed = now - self.updated_at
            # Refill lazily: compute how many tokens accumulated since last check
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.updated_at = now

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

Two details matter here:

1. time.monotonic() instead of time.time(). Wall-clock time jumps (NTP adjustments, manual clock changes) can make your limiter suddenly allow a huge burst or block everyone. Monotonic time only moves forward, which is what you want for measuring elapsed time.

2. Lazy refill. We don't run a background timer that adds tokens every second. We just compute how many tokens should have accumulated since the last request. This keeps the limiter cheap — the work happens only when a request arrives.

Wiring it into FastAPI

Dropping it into FastAPI takes a few lines:

from fastapi import FastAPI, HTTPException, Request

from token_bucket import TokenBucket

app = FastAPI()

# Allow 10 req/s sustained, with a burst of 30
bucket = TokenBucket(rate=10, capacity=30)


@app.middleware("http")
async def rate_limit(request: Request, call_next):
    client_id = request.client.host
    if not bucket.consume():
        raise HTTPException(
            status_code=429,
            detail="Too many requests",
            headers={"Retry-After": "1"},
        )
    return await call_next(request)

The Retry-After header is not optional polish — well-behaved clients read it and back off. Without it, clients just retry immediately and make the problem worse. A good limiter tells the caller when it's safe to try again.

Per-user buckets: the key matters more than the algorithm

Limiting by IP is the default, but it's often the wrong granularity. Users behind a corporate NAT share one IP — one heavy user can block everyone else at the office. Users on mobile networks cycle IPs constantly, so a per-IP limiter barely limits them at all.

What worked for us: limit by API key when the client is authenticated, fall back to IP for anonymous traffic. That way a single abusive integration gets throttled without collateral damage to unrelated users.

def client_id_for(request: Request) -> str:
    api_key = request.headers.get("X-Api-Key")
    if api_key:
        return f"key:{api_key}"
    return f"ip:{request.client.host}"

Making it work across multiple workers

The in-memory bucket above has a hidden trap: it's per-process. If you run four Uvicorn workers, each one has its own bucket with its own capacity — your effective limit becomes four times what you configured. For a small internal service that might be fine, but the moment you need a real guarantee, the bucket needs to live somewhere shared.

The pragmatic upgrade is Redis with a fixed window. It's not as smooth as a token bucket, but it's trivial to implement and it's a single shared store:

import time

import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)


def allow(client_id: str, limit: int = 100, window: int = 60) -> bool:
    key = f"rl:{client_id}:{int(time.time()) // window}"
    count = r.incr(key)
    if count == 1:
        r.expire(key, window)  # clean up after the window ends
    return count <= limit

Two gotchas with this version:

• The window boundary problem I mentioned earlier still exists — that's the trade-off you accept for simplicity.
• Set the TTL on the first increment (count == 1), not on every request, or you'll extend the window for active users indefinitely.

If you need distributed token-bucket semantics, the standard approach is a Lua script with sorted sets — but honestly, for most APIs the fixed window above is enough, and it's much easier to reason about.

Pitfalls I've hit so you don't have to

1. Rate limiting the wrong layer. Middleware that runs before auth can't key on the API key. Either parse the key in the middleware, or put the limiter inside the auth dependency.
2. Forgetting health checks and internal endpoints. You will eventually rate-limit your own health check and take down a load balancer's view of your service. Whitelist internal routes.
3. Too-large buckets. A capacity of 1000 with a rate of 10/s means the first 1000 requests in a burst all pass. If that's not what you want, keep the burst small and let the rate do the work.
4. No test coverage for the boundary. Write a test that fires capacity + 1 requests and asserts exactly one 429. It catches regressions the moment they happen.

When to use a library instead

For production, you'll probably reach for something like limits or slowapi (or your framework's built-in support). That's the right call — they handle edge cases I haven't covered here. But now you know what they're doing under the hood: the knobs they expose (rate, burst, storage backend) map directly to the concepts above. When a library's default settings feel wrong for your traffic, you'll know which parameter to touch.

Wrap-up

Rate limiting is one of those things that's invisible until it isn't. The good news is the core idea fits in ~40 lines: a token bucket, a lock, and a monotonic clock. Start with the in-memory version for a single-process service, switch to the Redis fixed window when you scale out, and key by user, not just by IP.

Your future self — and the person who gets paged at 2am — will thank you.