I Built 3 API Rate Limiters from Scratch — Here's the One I Kept

Last month, one of our API endpoints went from 200 req/min to 11,000 req/min in under five minutes. A new mobile app had a bug that retried failed requests infinitely. Our database did not appreciate the traffic spike.

That was the day I learned: a rate limiter isn't optional. It's the seatbelt you don't think about until you need it.

So I built three. Here's what I learned.

The Three Contenders

1. Fixed Window Counter

The simplest approach. Count requests in a time window, reset the counter when the window ends.

import time
from collections import defaultdict

class FixedWindowLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window = window_seconds
        self.counters = defaultdict(int)
        self.window_start = defaultdict(float)

    def allow(self, client_id: str) -> bool:
        now = time.time()
        ws = self.window_start.get(client_id, 0)

        # New window?
        if now - ws >= self.window:
            self.window_start[client_id] = now
            self.counters[client_id] = 0

        if self.counters[client_id] >= self.max_requests:
            return False

        self.counters[client_id] += 1
        return True

Pros: Dead simple. One counter per client.
Cons: The "edge burst" problem. A client that sends 100 requests right at the end of window A, then 100 more at the start of window B, gets 200 requests through in seconds — even with a limit of 100/minute.

I found this the hard way. My limiter reported "100/100 requests, all good!" while the database was on fire.

2. Sliding Window Log

This tracks the exact timestamp of every request and counts how many fall within the current window.

from collections import defaultdict, deque
import time

class SlidingWindowLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window = window_seconds
        self.logs = defaultdict(deque)

    def allow(self, client_id: str) -> bool:
        now = time.time()
        log = self.logs[client_id]
        cutoff = now - self.window

        # Remove old entries
        while log and log[0] < cutoff:
            log.popleft()

        if len(log) >= self.max_requests:
            return False

        log.append(now)
        return True

Pros: No burst problem. Precise.
Cons: Memory grows with request volume. A busy API with thousands of clients stores tons of timestamps. Also, calculating len(log) for every request adds up.

For our use case (~500 active clients, ~10K req/min total), this was fine. But I worried about scale.

3. Token Bucket

The elegant middle ground. Imagine a bucket that holds tokens. Each request consumes one token. Tokens refill at a steady rate. If the bucket is empty, requests are rejected.

import time
from collections import defaultdict

class TokenBucketLimiter:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate          # tokens per second
        self.capacity = capacity  # max burst size
        self.tokens = defaultdict(lambda: capacity)
        self.last_refill = defaultdict(float)

    def allow(self, client_id: str) -> bool:
        now = time.time()

        # Refill tokens based on elapsed time
        elapsed = now - self.last_refill.get(client_id, now)
        self.tokens[client_id] = min(
            self.capacity,
            self.tokens[client_id] + elapsed * self.rate
        )
        self.last_refill[client_id] = now

        if self.tokens[client_id] < 1:
            return False

        self.tokens[client_id] -= 1
        return True

Pros: Handles bursts naturally. Memory-efficient (one float per client).
Cons: Slightly harder to explain to your team on a whiteboard.

The Benchmark

I ran all three against a simulated traffic pattern: steady 5 req/s with random 50-request bursts every 30 seconds (limit set to 100 req/min):

Algorithm	Avg Latency	Burst Handling	Memory/Client
Fixed Window	0.001ms	Failed (double burst)	8 bytes
Sliding Window	0.08ms	Correct	~800 bytes
Token Bucket	0.002ms	Graceful	16 bytes

The Fixed Window was fast but wrong. Sliding Window was correct but slower (deque operations add up with high traffic). Token Bucket hit the sweet spot.

What I Kept

After three weeks of testing in production, I settled on Token Bucket. Here's why:

1. It handles bursts like a human would. A client that's been idle for 10 minutes should be able to send a few requests quickly — they've "saved up" their allowance. Fixed Window doesn't get this. Sliding Window doesn't reward patience.

2. It's stupidly memory-efficient. Two floats per client. That's it. When you have 5000+ clients, every byte counts.

3. It maps to real-world rate limits. Stripe, GitHub, and AWS all use token-bucket variants. When your API's behavior matches what developers already understand, you get fewer support tickets.

The Production Version

Here's the version that's been running for six months. It adds Redis persistence (so rate limits survive server restarts) and proper HTTP headers:

import redis
import time
from functools import wraps
from flask import request, jsonify

redis_client = redis.Redis(decode_responses=True)

class RedisTokenBucket:
    def __init__(self, key_prefix: str, rate: float, capacity: int):
        self.prefix = key_prefix
        self.rate = rate
        self.capacity = capacity

    def _keys(self, client_id: str):
        return (
            f"{self.prefix}:tokens:{client_id}",
            f"{self.prefix}:last:{client_id}"
        )

    def allow(self, client_id: str) -> tuple:
        """Returns (allowed, remaining_tokens, retry_after_seconds)"""
        tokens_key, last_key = self._keys(client_id)
        now = time.time()

        # Atomic check-and-update with Lua
        script = """
        local tokens = redis.call('get', KEYS[1]) or ARGV[1]
        local last = redis.call('get', KEYS[2]) or ARGV[2]
        local elapsed = tonumber(ARGV[2]) - tonumber(last)
        tokens = math.min(tonumber(ARGV[1]), tonumber(tokens) + elapsed * tonumber(ARGV[3]))

        if tokens < 1 then
            local retry_after = (1 - tokens) / tonumber(ARGV[3])
            return {0, 0, math.ceil(retry_after)}
        end

        tokens = tokens - 1
        redis.call('set', KEYS[1], tokens, 'EX', tonumber(ARGV[4]))
        redis.call('set', KEYS[2], ARGV[2])
        return {1, tokens, 0}
        """

        result = redis_client.eval(
            script, 2, tokens_key, last_key,
            self.capacity, now, self.rate, 3600
        )

        allowed, remaining, retry = result
        return bool(allowed), int(remaining), float(retry)

# Flask decorator
def rate_limit(max_per_minute: int = 60):
    limiter = RedisTokenBucket(
        key_prefix="rl",
        rate=max_per_minute / 60.0,
        capacity=max_per_minute
    )

    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            client_id = request.headers.get("X-API-Key", request.remote_addr)
            allowed, remaining, retry = limiter.allow(client_id)

            headers = {
                "X-RateLimit-Limit": str(max_per_minute),
                "X-RateLimit-Remaining": str(remaining),
                "X-RateLimit-Reset": str(int(time.time() + retry)),
            }

            if not allowed:
                headers["Retry-After"] = str(int(retry))
                return jsonify({"error": "Rate limit exceeded"}), 429, headers

            response = f(*args, **kwargs)
            for k, v in headers.items():
                response.headers[k] = v
            return response
        return wrapper
    return decorator

@app.route("/api/data")
@rate_limit(max_per_minute=60)
def get_data():
    return jsonify({"data": "here"})

Three Things I'd Do Differently

1. Start with Token Bucket, not Fixed Window. I built Fixed Window first because it was "simpler." It took a production incident to realize I needed to rewrite it. Cost: four hours of debugging and an angry Slack message from the infra team.

2. Add the HTTP headers from day one. X-RateLimit-Remaining and Retry-After aren't just nice-to-haves — they let client developers build smart retry logic. Without them, clients either retry blindly (making things worse) or give up (bad UX).

3. Don't roll your own Redis Lua scripts immediately. My first version used Python-level locking. It worked fine for months. Upgrade to Redis and Lua when you actually need it — premature optimization is the enemy of shipped code.

The Numbers That Matter

Six months after deploying the Token Bucket limiter:

• Zero rate-limit-related incidents
• 429 response rate: 0.3% of total requests
• Client-side retry success rate: 98.7% (because headers tell them when to try again)
• P99 latency added by rate limiter: 0.3ms

What About You?

What rate-limiting strategy are you using? Did I miss one you love? I'm especially curious about distributed rate limiting across multiple API gateways — that's our next challenge.

Drop your approach in the comments. I read every one.

---

Building API infrastructure that doesn't fall over. Follow for more war stories.