Redis-Backed Rate Limiter in FastAPI

Build a Token Bucket rate limiter with Redis and benchmark it with Locust.

Introduction

30 Second Summary

Every popular API faces the same challenge. Too many requests from a single user can overwhelm the server and ruin the experience for everyone else.

In this project, you will build a Token Bucket rate limiter backed by Redis for a FastAPI application, then load test it with Locust to measure how it performs under real traffic.

What You'll Build

A production-style rate limiting system that uses the Token Bucket algorithm with Redis as the state store, applied as middleware to a FastAPI app, and benchmarked with Locust for real performance metrics.

By the end of this project, you'll have:

  • A FastAPI application with multiple endpoints protected by your custom rate limiter that rejects excess requests with HTTP 429 responses.
  • A Token Bucket rate limiter class backed by Redis that enforces per-client limits and returns proper rate limit headers on every response.
  • A Locust load test script that simulates concurrent users and produces benchmark results showing requests per second, latency percentiles, and rejection rates.
  • Secret Mission: Add per-route rate limits with different tiers for expensive and cheap endpoints.

Are there any prerequisites?

You should have basic Python knowledge (functions, classes, async/await) and basic FastAPI knowledge (routes, requests, responses).

You also need Docker Desktop installed on your machine. We will use it to run Redis locally.

Before We Start

Before you can build a rate limiter, you need a working environment with all the right tools in place. This step sets up your project folder, installs the Python packages you need, and spins up a Redis server using Docker.

By the end of this step, you will have a clean workspace ready for coding. Your FastAPI, Redis, and Locust dependencies will be installed and verified.

In this step, get ready to:
  • Create your project folder and install Python dependencies.
  • Start a Redis server using Docker.
  • Create the project files you will build throughout this project.
Create your project folder and virtual environment
  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type the name of your code editor (VS Code or Cursor) and press Enter to open it.
  • Open the integrated terminal by pressing Ctrl+` (Windows) or Cmd+` (macOS).
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir rate-limiter-project
cd rate-limiter-project
  • Create a virtual environment by running:
python -m venv venv

What is a virtual environment?

A virtual environment is an isolated Python installation that keeps this project's packages separate from your system Python. This prevents version conflicts between different projects on your machine.

  • Activate your virtual environment:

macOS

  • Run this command to activate:
source venv/bin/activate

You should see (venv) appear at the start of your terminal prompt.

Windows

  • Run this command to activate:
venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt.

Don't see (venv) in your prompt?

On Windows, PowerShell may block script execution by default. Try running Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned and then re-run the activate command.

On macOS, make sure you typed source before the path. Without it, the command will not run correctly.

help me activate my Python virtual environment

  • Install the project dependencies by running:
pip install fastapi==0.136.3 uvicorn[standard] redis==5.3.1 locust==2.44.1

What does this install?

  • fastapi is the web framework you will build your API with.
  • uvicorn is the server that runs your FastAPI app.
  • redis is the Python client for communicating with your Redis server.
  • locust is the load testing tool you will use to benchmark your rate limiter.
  • Confirm all packages installed correctly by running:
python -c "import fastapi; import redis; import locust; print('All packages installed')"

You should see All packages installed printed in your terminal.

Seeing an install error?

Make sure your virtual environment is active (you should see (venv) in your prompt). If python is not recognized, try python3 instead.

help me fix my pip install error for fastapi, redis, or locust

Start Redis via Docker

Your rate limiter needs somewhere to store token counts for each client. Redis is a fast, in-memory data store that is perfect for this job.

  • Start a Redis container by running:
docker run -d --name redis -p 6379:6379 redis

What does this command do?

  • docker run starts a new container from the official Redis image.
  • -d runs it in the background so your terminal stays free.
  • --name redis gives the container a friendly name you can reference later.
  • -p 6379:6379 maps port 6379 on your machine to port 6379 inside the container so your Python code can connect.
  • Verify Redis is running by running:
docker exec -it redis redis-cli ping

You should see PONG printed in your terminal. This confirms Redis is running and accepting connections.

Not seeing PONG?

Make sure Docker Desktop is running before you execute the command. You can check by looking for the Docker whale icon in your system tray (Windows) or menu bar (macOS).

If you see a message about the container name already in use, remove the old one first by running docker rm redis and then re-run the docker run command.

help me get my Redis Docker container running

Create the project file structure

Your project needs four Python files. You will fill these in over the next few steps.

macOS

  • Create all four project files by running:
touch main.py rate_limiter.py middleware.py locustfile.py

Windows

  • Create all four project files by running:
New-Item -ItemType File -Path main.py, rate_limiter.py, middleware.py, locustfile.py
  • Confirm the files exist by running:
ls

You should see main.py, rate_limiter.py, middleware.py, and locustfile.py listed in the output.

Missing a file?

Double-check you ran the create command from inside the rate-limiter-project folder. If a file is missing, re-run the command for just that file (e.g. touch middleware.py on macOS).

help me create my project files for the rate limiter project

  • In your code editor, click File in the top menu bar.
  • Click Open Folder.
  • Select the rate-limiter-project folder on your Desktop, then click Open.

Your environment is ready. Next up, you will build the FastAPI application with endpoints that your rate limiter will protect.

Build the FastAPI App

Your environment is ready. Python packages are installed, Redis is running in Docker, and your project files are waiting to be filled.

Before you can rate limit anything, you need an API to protect. In this step, you'll build a FastAPI application with three endpoints that simulate different types of work. This gives you a realistic surface to apply rate limiting to in the next step.

In this step, get ready to:
  • Create a FastAPI application with three endpoints of varying cost.
  • Run the app and test all endpoints in your browser.
Create the FastAPI application with endpoints

Your API needs endpoints that behave differently so you can see how rate limiting affects each one. You'll create a lightweight health check, a standard data endpoint, and a heavy endpoint that simulates expensive work.

  • Add the following code to main.py to define your FastAPI app and its three endpoints:
import asyncio
from fastapi import FastAPI

app = FastAPI(title="Rate Limiter Demo")


@app.get("/api/health")
async def health():
    # Simple health check that returns immediately
    return {"status": "healthy"}


@app.get("/api/data")
async def get_data():
    # Simulates a normal data retrieval endpoint
    return {
        "message": "Here is your data",
        "items": ["item_1", "item_2", "item_3"],
    }


@app.get("/api/heavy")
async def heavy_operation():
    # Simulates an expensive operation (database query, external API call, etc.)
    await asyncio.sleep(0.5)
    return {"message": "Heavy operation complete", "result": 42}

What does this code do?

This file defines your entire API. Each endpoint simulates a different level of computational cost:

  • /api/health returns instantly. In production, load balancers use endpoints like this to check if a service is alive.
  • /api/data returns a sample JSON payload. This represents a typical data retrieval endpoint.
  • /api/heavy uses asyncio.sleep(0.5) to simulate an expensive operation like a database query or external API call. This is the kind of endpoint you'd want to rate limit aggressively.
  • Save main.py.
  • Start the development server by running this command in your terminal:
uvicorn main:app --reload
  • Open your browser and visit http://127.0.0.1:8000/api/health. You should see a JSON response: {"status": "healthy"}.

What does this command do?

Uvicorn is an ASGI server that runs your FastAPI application. The --reload flag tells it to restart automatically whenever you change your code. The main:app part means "look in main.py for the variable called app."

Don't see the JSON response?

  • Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
  • Check that you saved main.py before running the server. Unsaved changes won't be picked up.
  • If the terminal shows an import error, confirm your packages installed correctly by running pip list and checking that fastapi and uvicorn appear in the list.

Help me debug why my FastAPI server isn't starting.

✔️ Awesome, I've got everything!

Great. Make sure you've saved your file and your server is running before moving on.

ⓧ I'd like to double check the full code

Here's the complete main.py file:

import asyncio
from fastapi import FastAPI

app = FastAPI(title="Rate Limiter Demo")


@app.get("/api/health")
async def health():
    return {"status": "healthy"}


@app.get("/api/data")
async def get_data():
    return {
        "message": "Here is your data",
        "items": ["item_1", "item_2", "item_3"],
    }


@app.get("/api/heavy")
async def heavy_operation():
    await asyncio.sleep(0.5)
    return {"message": "Heavy operation complete", "result": 42}
Test all endpoints

With your server running, you can hit each endpoint to confirm they all respond correctly.

  • Visit http://127.0.0.1:8000/api/data in your browser. You should see a JSON object with a message and an items array.
  • Visit http://127.0.0.1:8000/api/heavy in your browser. This one takes about half a second to respond because it simulates expensive work.

Why does /api/heavy feel slower?

The half-second delay from asyncio.sleep(0.5) simulates a real-world expensive operation. In production, this could be a complex database query, a machine learning inference call, or a request to an external API. When you add rate limiting later, you'll want to be stricter with endpoints like this that consume more server resources.

All three endpoints should return valid JSON responses. The health and data endpoints respond instantly, while the heavy endpoint has a noticeable delay.

Your API is live and responding. Next up, you'll build the Token Bucket rate limiter that decides which requests get through and which get rejected.

Implement the Token Bucket Rate Limiter

Your FastAPI app serves three endpoints, but right now any client can hit them as often as they want. In production, uncontrolled traffic from a single client can overwhelm your server and ruin the experience for everyone else.

To fix this, you'll implement a rate limiter using the Token Bucket algorithm. Each client gets a "bucket" of tokens that refills over time. Every request costs one token, and when the bucket is empty, requests are denied until tokens refill.

In this step, get ready to:
  • Build the TokenBucket class that stores rate limit state in Redis.
  • Implement atomic token calculation logic using a Redis pipeline with optimistic locking.
Build the TokenBucket class

The Token Bucket algorithm works like a real bucket filling with water. Tokens drip in at a steady rate (the refill rate) up to a maximum (the capacity). Each request removes one token from the bucket.

  • Open rate_limiter.py in your editor and add the imports and class structure by pasting this code:
import time
import redis.asyncio as redis


class TokenBucket:
    def __init__(self, redis_client: redis.Redis, capacity: int = 10, refill_rate: float = 1.0):
        # Store the Redis connection and bucket configuration
        self.redis_client = redis_client
        self.capacity = capacity
        self.refill_rate = refill_rate

What does this code do?

import redis.asyncio as redis imports the async Redis client so your rate limiter can talk to Redis without blocking the event loop.

The TokenBucket class takes three parameters: a Redis client for storage, a capacity (maximum tokens in the bucket, defaulting to 10), and a refill_rate (tokens added per second, defaulting to 1.0).

You won't see a change yet. The next chunk adds the consume method that uses this class.

Add atomic token calculation logic

The core of the rate limiter is the consume method. On each request, it calculates how many tokens have refilled since the last check, then attempts to deduct one token atomically.

Why atomic operations?

If two requests arrive at the same time and both read the same token count, they might both think tokens are available and allow both through. Redis WATCH ensures that if another request modifies the bucket between your read and write, the operation fails and retries automatically.

  • Below the __init__ method (still inside the TokenBucket class), add the consume method by pasting this code:
    async def consume(self, key: str) -> tuple[bool, int, float]:
        """
        Attempt to consume a token for the given key.
        Returns: (allowed, remaining_tokens, retry_after_seconds)
        """
        now = time.time()
        redis_key = f"ratelimit:{key}"

        # Use a Redis pipeline with WATCH for atomic read-modify-write
        async with self.redis_client.pipeline(transaction=True) as pipe:
            try:
                # Watch the key so we detect changes from other requests
                await pipe.watch(redis_key)

                # Read current bucket state from Redis
                bucket_data = await pipe.hgetall(redis_key)

                if bucket_data:
                    tokens = float(bucket_data[b"tokens"])
                    last_refill = float(bucket_data[b"last_refill"])
                else:
                    # First request from this client - start with a full bucket
                    tokens = float(self.capacity)
                    last_refill = now

                # Refill tokens based on elapsed time (capped at capacity)
                elapsed = now - last_refill
                tokens = min(self.capacity, tokens + elapsed * self.refill_rate)
                last_refill = now

                # Try to consume one token
                if tokens >= 1:
                    tokens -= 1
                    allowed = True
                    retry_after = 0.0
                else:
                    # No tokens available - calculate wait time
                    allowed = False
                    retry_after = (1 - tokens) / self.refill_rate

                # Write updated state back to Redis atomically
                pipe.multi()
                pipe.hset(redis_key, mapping={
                    "tokens": str(tokens),
                    "last_refill": str(last_refill),
                })
                pipe.expire(redis_key, 3600)
                await pipe.execute()

                return allowed, int(tokens), retry_after

            except redis.WatchError:
                # Another request modified the key - retry
                return await self.consume(key)

What does this code do?

  • pipe.watch(redis_key) tells Redis to monitor this key for changes while we calculate.
  • pipe.hgetall(redis_key) reads the current token count and last refill timestamp from a Redis hash.
  • The refill calculation adds tokens based on elapsed time, capped at self.capacity so the bucket never overflows.
  • pipe.multi() starts the atomic write. If another request changed the key since our watch, Redis raises WatchError and the method retries by calling itself recursively.
  • pipe.expire(redis_key, 3600) auto-deletes inactive buckets after one hour to prevent stale data buildup.
  • Save rate_limiter.py.
  • Verify the class loads without errors by running this command in your terminal:
python -c "from rate_limiter import TokenBucket; print('TokenBucket class loaded successfully')"

You should see TokenBucket class loaded successfully printed in your terminal.

Seeing an ImportError or SyntaxError?

  • Make sure the consume method is indented inside the TokenBucket class (one level of indentation from the class definition).
  • Check that your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
  • Confirm redis is installed by running pip list | findstr redis and checking for redis 5.3.1.

Help me debug my TokenBucket import error

✔️ Awesome, I've got everything!

Great. Double check you've saved rate_limiter.py before moving on.

ⓧ I'd like to double check the full code

import time
import redis.asyncio as redis


class TokenBucket:
    def __init__(self, redis_client: redis.Redis, capacity: int = 10, refill_rate: float = 1.0):
        self.redis_client = redis_client
        self.capacity = capacity
        self.refill_rate = refill_rate

    async def consume(self, key: str) -> tuple[bool, int, float]:
        """
        Attempt to consume a token for the given key.
        Returns: (allowed, remaining_tokens, retry_after_seconds)
        """
        now = time.time()
        redis_key = f"ratelimit:{key}"

        async with self.redis_client.pipeline(transaction=True) as pipe:
            try:
                await pipe.watch(redis_key)

                bucket_data = await pipe.hgetall(redis_key)

                if bucket_data:
                    tokens = float(bucket_data[b"tokens"])
                    last_refill = float(bucket_data[b"last_refill"])
                else:
                    tokens = float(self.capacity)
                    last_refill = now

                elapsed = now - last_refill
                tokens = min(self.capacity, tokens + elapsed * self.refill_rate)
                last_refill = now

                if tokens >= 1:
                    tokens -= 1
                    allowed = True
                    retry_after = 0.0
                else:
                    allowed = False
                    retry_after = (1 - tokens) / self.refill_rate

                pipe.multi()
                pipe.hset(redis_key, mapping={
                    "tokens": str(tokens),
                    "last_refill": str(last_refill),
                })
                pipe.expire(redis_key, 3600)
                await pipe.execute()

                return allowed, int(tokens), retry_after

            except redis.WatchError:
                return await self.consume(key)

Your rate limiter is ready. Next up, you'll wire it into FastAPI as middleware so every incoming request passes through the token bucket before reaching your endpoints.

Wire the Rate Limiter into FastAPI as Middleware

You have a working Token Bucket algorithm that tracks tokens in Redis. But it's sitting in its own file, disconnected from incoming HTTP requests.

In this step, you'll write middleware that intercepts every request, checks the token bucket, and either lets the request through or rejects it with a proper HTTP 429 response. You'll also attach rate limit headers so clients always know how much quota they have left.

In this step, get ready to:
  • Create the rate limit middleware function.
  • Wire Redis, the token bucket, and the middleware together in your FastAPI app.
  • Verify that rate limit headers appear on every response.
Create the middleware in middleware.py

Middleware sits between every incoming request and your route handlers. Your middleware needs to do two things: reject requests that exceed the rate limit, and attach informational headers to every response that gets through.

  • In your code editor, open middleware.py and add the following code:
from fastapi import Request
from fastapi.responses import JSONResponse
from rate_limiter import TokenBucket


async def rate_limit_middleware(request: Request, call_next, token_bucket: TokenBucket):
    # Identify the client by their IP address
    client_ip = request.client.host
    # Check if this client has tokens remaining
    allowed, remaining, retry_after = await token_bucket.consume(client_ip)

    if not allowed:
        # Reject with 429 and tell the client when to retry
        return JSONResponse(
            status_code=429,
            content={"detail": "Rate limit exceeded. Try again later."},
            headers={
                "Retry-After": str(int(retry_after) + 1),
                "X-RateLimit-Limit": str(token_bucket.capacity),
                "X-RateLimit-Remaining": "0",
            },
        )

    # Request is allowed - forward to the route handler
    response = await call_next(request)
    # Attach rate limit info so clients can monitor their quota
    response.headers["X-RateLimit-Limit"] = str(token_bucket.capacity)
    response.headers["X-RateLimit-Remaining"] = str(remaining)
    return response

What does this code do?

This function acts as a gatekeeper for every HTTP request your app receives.

  • request.client.host extracts the client's IP address, which the token bucket uses as the unique key per client.
  • If allowed is False, the middleware short-circuits the request with an HTTP 429 response. The Retry-After header tells the client how many seconds to wait before trying again.
  • If allowed, call_next(request) passes the request to the actual route handler. The response then gets X-RateLimit-Limit and X-RateLimit-Remaining headers attached so the client can see how much quota is left.
  • Save middleware.py.

Why include rate limit headers on successful responses?

Well-designed APIs tell clients about their remaining quota on every response, not just when they're blocked. This lets clients monitor their usage proactively and slow down before hitting the limit. Services like GitHub, Stripe, and Discord all follow this pattern.

Connect everything in main.py

The middleware function is ready, but it needs a Redis connection and a TokenBucket instance to work with. You'll wire all the pieces together in main.py using FastAPI's startup and shutdown lifecycle events.

  • Replace the entire contents of main.py with the following code:
import asyncio
import redis.asyncio as redis
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from rate_limiter import TokenBucket
from middleware import rate_limit_middleware

app = FastAPI(title="Rate Limiter Demo")

# Global references to Redis and the token bucket
redis_client: redis.Redis = None
token_bucket: TokenBucket = None


@app.on_event("startup")
async def startup():
    # Connect to Redis when the app starts
    global redis_client, token_bucket
    redis_client = redis.Redis(host="localhost", port=6379, db=0)
    token_bucket = TokenBucket(redis_client=redis_client, capacity=10, refill_rate=1.0)


@app.on_event("shutdown")
async def shutdown():
    # Close the Redis connection cleanly
    await redis_client.aclose()


@app.middleware("http")
async def apply_rate_limit(request: Request, call_next):
    return await rate_limit_middleware(request, call_next, token_bucket)


@app.get("/api/health")
async def health():
    return {"status": "healthy"}


@app.get("/api/data")
async def get_data():
    return {
        "message": "Here is your data",
        "items": ["item_1", "item_2", "item_3"],
    }


@app.get("/api/heavy")
async def heavy_operation():
    await asyncio.sleep(0.5)
    return {"message": "Heavy operation complete", "result": 42}

What does this code do?

  • The startup event creates the Redis connection and a TokenBucket with 10 tokens max capacity and 1 token refilled per second. This runs once when the app starts.
  • The shutdown event calls aclose() to cleanly disconnect from Redis when the app stops.
  • The @app.middleware("http") decorator registers apply_rate_limit as middleware that runs on every HTTP request. It delegates to your rate_limit_middleware function, passing along the token bucket.
  • Save main.py.
  • Start the app by running this command in your terminal:
uvicorn main:app --reload
  • Open your browser and visit http://127.0.0.1:8000/api/data repeatedly. After about 10 rapid refreshes, you should see a JSON response with the message Rate limit exceeded. Try again later. instead of your data.

Still seeing successful responses after 10+ refreshes?

The token bucket refills at 1 token per second. If you're refreshing slowly (once per second or slower), tokens refill as fast as you use them. Try clicking refresh as fast as you can to exhaust the bucket.

If you see a connection error instead, make sure your Redis container is still running. Check with docker exec -it redis redis-cli ping in a separate terminal.

Help me debug my rate limiter middleware

Verify rate limit headers

Your rate limiter is rejecting excess traffic. Now confirm that the response headers are present so clients can programmatically track their remaining quota.

  • Open a new terminal and run this command to inspect the full response headers:
curl -v http://127.0.0.1:8000/api/data

What should I see?

In the response headers (lines starting with <), you should see X-RateLimit-Limit: 10 and X-RateLimit-Remaining with a number showing how many tokens you have left.

Run the same curl command several more times rapidly. Watch X-RateLimit-Remaining decrease with each request. Once it hits 0, the next request returns HTTP 429 with a Retry-After header.

  • Run the curl -v command at least 11 times rapidly to trigger a 429 response.
  • Confirm that the 429 response includes the Retry-After header showing how many seconds to wait.

Not seeing the rate limit headers?

Make sure uvicorn is still running in your other terminal. The --reload flag automatically restarts the server when you save files, so it should have picked up your changes.

If headers appear but with unexpected values, wait 10 seconds for the bucket to fully refill, then try the rapid curl test again from a fresh state.

Help me debug my middleware headers

✔️ Awesome, I've got everything!

Great. Double check you've saved both middleware.py and main.py.

ⓧ I'd like to double check the full code

Here is the complete middleware.py:

from fastapi import Request
from fastapi.responses import JSONResponse
from rate_limiter import TokenBucket


async def rate_limit_middleware(request: Request, call_next, token_bucket: TokenBucket):
    client_ip = request.client.host
    allowed, remaining, retry_after = await token_bucket.consume(client_ip)

    if not allowed:
        return JSONResponse(
            status_code=429,
            content={"detail": "Rate limit exceeded. Try again later."},
            headers={
                "Retry-After": str(int(retry_after) + 1),
                "X-RateLimit-Limit": str(token_bucket.capacity),
                "X-RateLimit-Remaining": "0",
            },
        )

    response = await call_next(request)
    response.headers["X-RateLimit-Limit"] = str(token_bucket.capacity)
    response.headers["X-RateLimit-Remaining"] = str(remaining)
    return response

Here is the complete main.py:

import asyncio
import redis.asyncio as redis
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from rate_limiter import TokenBucket
from middleware import rate_limit_middleware

app = FastAPI(title="Rate Limiter Demo")

redis_client: redis.Redis = None
token_bucket: TokenBucket = None


@app.on_event("startup")
async def startup():
    global redis_client, token_bucket
    redis_client = redis.Redis(host="localhost", port=6379, db=0)
    token_bucket = TokenBucket(redis_client=redis_client, capacity=10, refill_rate=1.0)


@app.on_event("shutdown")
async def shutdown():
    await redis_client.aclose()


@app.middleware("http")
async def apply_rate_limit(request: Request, call_next):
    return await rate_limit_middleware(request, call_next, token_bucket)


@app.get("/api/health")
async def health():
    return {"status": "healthy"}


@app.get("/api/data")
async def get_data():
    return {
        "message": "Here is your data",
        "items": ["item_1", "item_2", "item_3"],
    }


@app.get("/api/heavy")
async def heavy_operation():
    await asyncio.sleep(0.5)
    return {"message": "Heavy operation complete", "result": 42}

Your API now protects itself from abuse. Every client gets 10 requests before hitting the limit, and the headers give them full visibility into their remaining quota. Next up, you'll throw real load at this system with Locust to see how it performs under pressure.

Benchmark with Locust

Your rate limiter is working. It correctly rejects requests that exceed the token bucket's capacity and returns proper HTTP 429 responses with headers.

But how well does it perform under real traffic? In this step, you'll write a Locust load testing script that simulates concurrent users hitting your API. You'll then run a benchmark to measure throughput, latency, and rejection rate.

In this step, get ready to:
  • Write a Locust load test script that simulates concurrent users.
  • Run a benchmark with 50 users and observe real-time metrics.
  • Scale up to 200 users and analyze throughput, latency, and rejection rates.
Write the Locust test script

Locust tests are Python scripts that define simulated user behaviour. Each user class specifies which endpoints to hit and how often.

  • Open locustfile.py in your editor and add the following code:
from locust import HttpUser, task, between


class APIUser(HttpUser):
    # Wait between 0.1 and 0.5 seconds between requests
    wait_time = between(0.1, 0.5)

    @task(1)
    def check_health(self):
        # Low weight - light health check
        self.client.get("/api/health")

    @task(5)
    def get_data(self):
        # Highest weight - most common user action
        self.client.get("/api/data")

    @task(2)
    def heavy_request(self):
        # Medium weight - simulates expensive operations
        self.client.get("/api/heavy")

What does this code do?

The APIUser class extends HttpUser, which tells Locust to make HTTP requests as this user.

  • wait_time = between(0.1, 0.5) makes each simulated user pause 0.1 to 0.5 seconds between requests. This keeps the load aggressive for testing.
  • The @task(weight) decorator controls how often each method runs. A weight of 5 on get_data means it fires 5x more often than check_health (weight 1).
  • self.client.get(...) sends actual HTTP GET requests to your running FastAPI app.
  • Save locustfile.py.

Seeing import errors?

  • Make sure your virtual environment is activated. You should see (venv) in your terminal prompt.
  • Confirm Locust is installed by running pip show locust in your terminal.

Help me fix my Locust import errors.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file.

ⓧ I'd like to double check the full code

from locust import HttpUser, task, between


class APIUser(HttpUser):
    wait_time = between(0.1, 0.5)

    @task(1)
    def check_health(self):
        self.client.get("/api/health")

    @task(5)
    def get_data(self):
        self.client.get("/api/data")

    @task(2)
    def heavy_request(self):
        self.client.get("/api/heavy")
Run the Locust benchmark

With the script ready, you can now point Locust at your running API and watch it generate real traffic.

  • Open a new terminal window. Your FastAPI app should still be running in the first terminal via uvicorn.
  • Make sure you are inside your rate-limiter-project folder and your virtual environment is activated.
  • Start Locust by running this command:
locust -f locustfile.py --host=http://127.0.0.1:8000

What does this command do?

  • -f locustfile.py tells Locust which test script to load.
  • --host=http://127.0.0.1:8000 sets the target server. This is where your FastAPI app is running.
  • Locust starts a web UI on port 8089 where you can configure and monitor your load test.
  • Open your browser and go to http://localhost:8089.
  • In the Number of users field, enter 50.
  • In the Ramp up (users/second) field, enter 10.
  • Click Start to begin the load test.
  • Let the test run for approximately 60 seconds. Watch the statistics table update in real time.

You should see the statistics table showing requests to /api/health, /api/data, and /api/heavy. The Failure Count column shows 429 rejections from your rate limiter.

What should I see?

The Locust web UI shows a table with columns including Requests/s, Median Response Time, Failure Count, and response time percentiles. Under the Charts tab, you can see requests per second, response times, and number of running users over time.

If your failure count is climbing, that means your rate limiter is working. The 429 responses are Locust reporting "failures" because the status code is not 2xx.

Not seeing any failures?

  • Make sure your FastAPI app is still running in the other terminal. If uvicorn crashed, restart it.
  • Check that the --host value matches the port your app is running on (8000 by default).
  • Confirm the Redis container is still running with docker ps. If it stopped, restart it with docker start redis.

Help me troubleshoot why my Locust test isn't showing 429 failures.

Analyze benchmark results

The numbers in the Locust UI tell a story about your rate limiter's behaviour under pressure. Understanding them is the difference between "it works" and "I know how it performs."

  • Look at the Requests/s column in the Aggregated row. This is your API's total throughput.
  • Note the Median Response Time and the 95th percentile. The median tells you the typical experience. The 95th percentile tells you the worst-case for most users.
  • Calculate the failure percentage. Divide the Failure Count by the total Request Count. This is your rejection rate.

What do these metrics mean?

  • Requests/s (throughput): how many requests your API handles per second. Higher is better for allowed traffic.
  • Median Response Time: the response time for 50% of requests. Low median means fast for most users.
  • 95th percentile: the response time that 95% of requests fall under. A big gap between median and 95th percentile means some requests experience much higher latency.
  • Failure percentage (429 rejections): this is your rate limiter doing its job. A high rejection rate with 50 users confirms that your 10-token bucket gets exhausted quickly under load.
  • Click Stop in the Locust UI to end the current test.
  • Click New test in the top navigation.
  • This time, enter 200 in the Number of users field.
  • Keep the ramp up at 10 and click Start.
  • Let this run for 60 seconds and compare the results to your first test.

With 200 users, you should see a much higher failure rate. The rate limiter allows the same 10 requests per second regardless of how many users are trying. More users means more rejections.

Why more users means more rejections

Your token bucket refills at 1 token per second with a max capacity of 10. With 200 simulated users sending requests every 0.1 to 0.5 seconds, demand far outstrips the 10 tokens available. The rejection rate climbs because the bucket empties almost instantly after each refill.

In production, this is exactly what you want. Your API stays responsive for allowed traffic while excess load gets cleanly rejected with a Retry-After header telling clients when to try again.

  • Click Stop to end the test when you're finished observing.

Secret mission

Add Per-Route Rate Limits

Right now, every endpoint shares the same 10-token bucket. In production, an expensive database query and a lightweight health check should not compete for the same quota. Your mission: give each route its own rate limit tier and prove the difference under load.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • A Docker container named redis running on port 6379.
  • Local project folder rate-limiter-project containing your Python files and virtual environment.

Keep everything running

No action needed. Choose this if you're still actively building or experimenting with your rate limiter.

  • Your Redis container continues running in the background. It uses minimal memory and CPU.
  • Your project files remain in rate-limiter-project on your Desktop.
  • To restart your FastAPI app later, activate your virtual environment and run uvicorn main:app --reload from inside the project folder.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your files and container data stay intact.

  • Stop the Redis container by running:
docker stop redis
  • Stop your FastAPI server by pressing Ctrl+C in the terminal where it's running.
  • When you're ready to come back, restart Redis by running:
docker start redis

Delete - I don't want to use this again

Remove all project resources and start fresh. This is permanent.

  • Stop and remove the Redis container by running:
docker stop redis
docker rm redis
  • Delete the project folder. On macOS/Linux, run:
cd ~/Desktop
rm -rf rate-limiter-project
  • On Windows, run:
cd ~/Desktop
rmdir /s /q rate-limiter-project
  • This removes all your Python files, the virtual environment, and installed packages.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-style rate limiting system from scratch, wired it into a FastAPI application, and proved it works under real load.

You've learned how to:

  • Build a Token Bucket rate limiter backed by Redis atomic operations, ensuring correctness under concurrency.
  • Create FastAPI middleware that enforces per-client rate limits with proper HTTP headers (X-RateLimit-Limit, X-RateLimit-Remaining, Retry-After).
  • Write Locust load tests and analyze real benchmark metrics including requests per second, latency percentiles, and 429 rejection rates.
  • Secret Mission: Implemented per-route rate limiting with different tiers for different endpoints.

Ready to quiz yourself?