Build a Rate Limiter with FastAPI

Build a production-grade async rate limiter with two algorithms and FastAPI.

Introduction

30 Second Summary

Every time you've been told "slow down" by an API or seen a 429 error, a rate limiter was behind it. These invisible gatekeepers protect every major service you use daily, from Stripe's payment endpoints to GitHub's API.

In this project, you will build a production-grade rate limiting library in Python with two swappable algorithms, integrate it into a FastAPI application as middleware, and test it under concurrent load with pytest.

What You'll Build

A fully async rate limiting library supporting two algorithms via the Strategy pattern, integrated into FastAPI middleware with proper HTTP 429 responses and tested with concurrent async requests.

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

  • Two rate limiting algorithms (Token Bucket and Sliding Window Log) behind a clean Protocol interface that makes them swappable without changing middleware code.
  • A FastAPI middleware that applies rate limiting per-client based on IP and returns standards-compliant 429 responses with Retry-After headers.
  • A pytest test suite validating concurrent access with asyncio.TaskGroup and httpx.AsyncClient.
  • Secret Mission: Add a decorator-based rate limiter that can be applied per-endpoint with different limits.

Are there any prerequisites?

You should be comfortable with Python classes, async/await basics, and HTTP concepts. You don't need prior experience with FastAPI or rate limiting algorithms.

Define the Rate Limiter Protocol and Token Bucket

Your project structure is ready with dependencies installed. Now you'll write the core logic that makes a rate limiter work.

Before jumping into algorithm code, you'll define a Protocol class that acts as a contract for any rate limiter implementation. Then you'll build your first algorithm: the Token Bucket.

In this step, get ready to:
  • Create the RateLimiter Protocol class with type-hinted async methods.
  • Implement TokenBucketLimiter with configurable rate and capacity.
  • Verify the token bucket refills correctly over time.
Create the rate limiter Protocol

A Protocol in Python defines what methods a class must have, without forcing it to inherit from a base class. Any class that implements the right method signatures automatically satisfies the protocol.

  • Create a new file called src/protocol.py inside your rate-limiter project folder.
  • Add the following code to define the RateLimiter interface:
from typing import Protocol


class RateLimiter(Protocol):
    async def is_allowed(self, key: str) -> bool: ...
    async def time_until_available(self, key: str) -> float: ...
    async def reset(self, key: str) -> None: ...

What does this code do?

This Protocol defines the contract that every rate limiter algorithm must satisfy. Any class with these three async methods is automatically considered a valid RateLimiter, no inheritance required.

  • is_allowed(key) checks whether a given client (identified by key) can make a request right now.
  • time_until_available(key) returns how many seconds the client must wait before their next request will be allowed.
  • reset(key) clears all state for a client, restoring their full allowance.
  • Save src/protocol.py.

Why Protocol instead of an abstract base class?

Python's Protocol uses structural typing (also called static duck typing). A class satisfies the protocol by having the right methods. It never needs to inherit from RateLimiter directly.

This is more Pythonic than abstract base classes because it avoids forced inheritance. It also makes testing easier since any object with the right shape works as a drop-in replacement.

✔️ Awesome, I've got everything!

Great. Make sure you've saved src/protocol.py before moving on.

ⓧ I'd like to double check the full code

from typing import Protocol


class RateLimiter(Protocol):
    async def is_allowed(self, key: str) -> bool: ...
    async def time_until_available(self, key: str) -> float: ...
    async def reset(self, key: str) -> None: ...
Implement the Token Bucket algorithm

The Token Bucket algorithm works like a bucket that fills up with tokens at a steady rate. Each request costs one token. If the bucket is empty, the request is denied.

This approach handles bursty traffic well because clients can spend their accumulated tokens in quick succession, up to the bucket's maximum capacity.

  • Create a new file called src/token_bucket.py inside your src/ directory.
  • Add the full Token Bucket implementation:
import asyncio
import time


class TokenBucketLimiter:
    def __init__(self, rate: float, capacity: int) -> None:
        self._rate = rate
        self._capacity = capacity
        self._buckets: dict[str, dict[str, float]] = {}
        self._lock = asyncio.Lock()

    def _get_bucket(self, key: str) -> dict[str, float]:
        if key not in self._buckets:
            self._buckets[key] = {
                "tokens": float(self._capacity),
                "last_refill": time.monotonic(),
            }
        return self._buckets[key]

    def _refill(self, bucket: dict[str, float]) -> None:
        now = time.monotonic()
        elapsed = now - bucket["last_refill"]
        bucket["tokens"] = min(
            self._capacity, bucket["tokens"] + elapsed * self._rate
        )
        bucket["last_refill"] = now

    async def is_allowed(self, key: str) -> bool:
        async with self._lock:
            bucket = self._get_bucket(key)
            self._refill(bucket)
            if bucket["tokens"] >= 1.0:
                bucket["tokens"] -= 1.0
                return True
            return False

    async def time_until_available(self, key: str) -> float:
        async with self._lock:
            bucket = self._get_bucket(key)
            self._refill(bucket)
            if bucket["tokens"] >= 1.0:
                return 0.0
            tokens_needed = 1.0 - bucket["tokens"]
            return tokens_needed / self._rate

    async def reset(self, key: str) -> None:
        async with self._lock:
            self._buckets.pop(key, None)

What does this code do?

The TokenBucketLimiter manages a separate bucket for each client key. Each bucket tracks how many tokens remain and when it was last refilled.

  • rate controls how fast tokens regenerate (tokens per second). capacity sets the maximum burst size.
  • _refill() calculates elapsed time since the last check and adds proportional tokens, capped at capacity.
  • _lock is an asyncio.Lock that prevents race conditions when multiple async tasks check the same bucket simultaneously.
  • time.monotonic() provides a clock that never jumps backward, making elapsed-time calculations reliable.
  • Save src/token_bucket.py.

Seeing a red underline on dict[str, dict[str, float]]?

This syntax requires Python 3.9 or higher. Since the project uses Python 3.13, this should work. If your editor still flags it, make sure your IDE is pointing at the correct Python interpreter from your project's virtual environment.

Help me fix this type annotation error

✔️ Awesome, I've got everything!

Great. Make sure you've saved src/token_bucket.py before continuing.

ⓧ I'd like to double check the full code

import asyncio
import time


class TokenBucketLimiter:
    def __init__(self, rate: float, capacity: int) -> None:
        self._rate = rate
        self._capacity = capacity
        self._buckets: dict[str, dict[str, float]] = {}
        self._lock = asyncio.Lock()

    def _get_bucket(self, key: str) -> dict[str, float]:
        if key not in self._buckets:
            self._buckets[key] = {
                "tokens": float(self._capacity),
                "last_refill": time.monotonic(),
            }
        return self._buckets[key]

    def _refill(self, bucket: dict[str, float]) -> None:
        now = time.monotonic()
        elapsed = now - bucket["last_refill"]
        bucket["tokens"] = min(
            self._capacity, bucket["tokens"] + elapsed * self._rate
        )
        bucket["last_refill"] = now

    async def is_allowed(self, key: str) -> bool:
        async with self._lock:
            bucket = self._get_bucket(key)
            self._refill(bucket)
            if bucket["tokens"] >= 1.0:
                bucket["tokens"] -= 1.0
                return True
            return False

    async def time_until_available(self, key: str) -> float:
        async with self._lock:
            bucket = self._get_bucket(key)
            self._refill(bucket)
            if bucket["tokens"] >= 1.0:
                return 0.0
            tokens_needed = 1.0 - bucket["tokens"]
            return tokens_needed / self._rate

    async def reset(self, key: str) -> None:
        async with self._lock:
            self._buckets.pop(key, None)
Verify the token bucket refills over time

Before building more, confirm that the token bucket actually works. You'll write a quick script that exhausts the bucket, waits, and checks that tokens refill.

  • Create a file called test_manual.py in the root of your rate-limiter folder (next to pyproject.toml):
import asyncio
from src.token_bucket import TokenBucketLimiter


async def main():
    # Create a bucket: 2 tokens/sec refill rate, max 3 tokens
    limiter = TokenBucketLimiter(rate=2.0, capacity=3)

    # Use all 3 tokens plus attempt a 4th
    for i in range(4):
        result = await limiter.is_allowed("test-user")
        print(f"Request {i + 1}: {'allowed' if result else 'blocked'}")

    # Wait for tokens to refill
    print("\nWaiting 1 second for refill...")
    await asyncio.sleep(1.0)

    # Try again after waiting
    result = await limiter.is_allowed("test-user")
    print(f"After 1s wait: {'allowed' if result else 'blocked'}")


asyncio.run(main())

What does this script do?

This script creates a bucket with capacity 3 and a refill rate of 2 tokens per second. It fires 4 requests immediately (the 4th should be blocked), waits 1 second for refill, then tries again.

If the algorithm works, you'll see 3 allowed, 1 blocked, then 1 allowed after the wait.

  • Save test_manual.py and run it with this command:
uv run python test_manual.py

You should see this output:

Request 1: allowed
Request 2: allowed
Request 3: allowed
Request 4: blocked

Waiting 1 second for refill...
After 1s wait: allowed

The token bucket is working. Requests 1-3 consumed the initial 3 tokens, request 4 was blocked, and after waiting 1 second the bucket refilled enough for another request.

Seeing ModuleNotFoundError: No module named 'src'?

Make sure you're running the command from the rate-limiter directory (the one containing pyproject.toml). The uv run command resolves imports relative to the project root.

Help me fix this import error

  • You can delete test_manual.py now. It was only for verification.

Your first rate limiting algorithm is complete and verified. Next up, you'll implement a second algorithm (Sliding Window Log) so you can compare how they handle traffic differently.

Implement the Sliding Window Log Algorithm

Your Token Bucket limiter handles bursty traffic gracefully by refilling tokens over time. But what if you need strict, precise enforcement where exactly N requests are allowed per time window, no exceptions?

That's where the Sliding Window Log algorithm comes in. Instead of tokens, it records the exact timestamp of every request and counts how many fall within the current window. In this step, you'll build this second algorithm and compare both side by side.

In this step, get ready to:
  • Implement the SlidingWindowLogLimiter class using a deque to track timestamps.
  • Add the time_until_available method for Retry-After header support.
  • Run both algorithms side by side to observe their behavioral differences.
Implement the SlidingWindowLogLimiter class

The Sliding Window Log algorithm keeps a log of every request timestamp. When a new request comes in, it discards expired entries and checks whether the remaining count is under the limit.

  • Create a new file called src/sliding_window.py in VS Code.
  • Add the class foundation with its imports, constructor, and private helpers by pasting this code:
import asyncio
import time
from collections import deque


class SlidingWindowLogLimiter:
    def __init__(self, max_requests: int, window_seconds: float) -> None:
        self._max_requests = max_requests
        self._window_seconds = window_seconds
        self._logs: dict[str, deque[float]] = {}
        self._lock = asyncio.Lock()

    def _get_log(self, key: str) -> deque[float]:
        if key not in self._logs:
            self._logs[key] = deque()
        return self._logs[key]

    def _cleanup(self, log: deque[float], now: float) -> None:
        cutoff = now - self._window_seconds
        while log and log[0] <= cutoff:
            log.popleft()

What does this code do?

The constructor takes two config values: max_requests (how many requests are allowed) and window_seconds (the time window to count within). The _logs dictionary maps each client key to a deque of timestamps.

The _cleanup method removes timestamps older than the window. Since the deque is ordered chronologically, it just pops from the left until all remaining entries are within the window.

Now add the core is_allowed method that decides whether a request should pass or be rejected.

  • Add this method inside the SlidingWindowLogLimiter class, below _cleanup:
    async def is_allowed(self, key: str) -> bool:
        async with self._lock:
            now = time.monotonic()
            log = self._get_log(key)
            self._cleanup(log, now)
            if len(log) < self._max_requests:
                log.append(now)
                return True
            return False

What does this method do?

Under the async lock, it grabs the current time, cleans out expired timestamps, then checks if the log has room. If the count is below max_requests, it records the new timestamp and returns True. Otherwise it returns False.

This is the key difference from the token bucket. There are no partial refills. Either you have room in the window or you don't.

Why use a deque instead of a list?

A deque (double-ended queue) supports O(1) appends and pops from both ends. Since cleanup always removes from the left and new entries append to the right, a deque is the ideal data structure. A regular list would be O(n) for left-side removals.

Add time_until_available and reset

The time_until_available method tells a rejected client how long to wait before retrying. Your Token Bucket already has this from the previous step. Now you'll add it to the Sliding Window.

  • Add these two methods below is_allowed in src/sliding_window.py:
    async def time_until_available(self, key: str) -> float:
        async with self._lock:
            now = time.monotonic()
            log = self._get_log(key)
            self._cleanup(log, now)
            if len(log) < self._max_requests:
                return 0.0
            oldest = log[0]
            return (oldest + self._window_seconds) - now

    async def reset(self, key: str) -> None:
        async with self._lock:
            self._logs.pop(key, None)

What does time_until_available calculate?

If the log is full, the client must wait until the oldest entry expires out of the window. The calculation (oldest + window_seconds) - now gives exactly that remaining time. This value becomes the Retry-After header in HTTP 429 responses.

The reset method simply removes the key's log entirely, restoring full capacity for that client.

  • Save src/sliding_window.py.

Seeing an indentation error?

All the async methods must be indented one level inside the class body (4 spaces). If you see IndentationError, check that async def is_allowed, async def time_until_available, and async def reset are all at the same indent level as _get_log and _cleanup.

Help me fix an indentation error in my SlidingWindowLogLimiter class.

✔️ Awesome, I've got everything!

Great. Make sure you've saved src/sliding_window.py before moving on.

ⓧ I'd like to double check the full code

import asyncio
import time
from collections import deque


class SlidingWindowLogLimiter:
    def __init__(self, max_requests: int, window_seconds: float) -> None:
        self._max_requests = max_requests
        self._window_seconds = window_seconds
        self._logs: dict[str, deque[float]] = {}
        self._lock = asyncio.Lock()

    def _get_log(self, key: str) -> deque[float]:
        if key not in self._logs:
            self._logs[key] = deque()
        return self._logs[key]

    def _cleanup(self, log: deque[float], now: float) -> None:
        cutoff = now - self._window_seconds
        while log and log[0] <= cutoff:
            log.popleft()

    async def is_allowed(self, key: str) -> bool:
        async with self._lock:
            now = time.monotonic()
            log = self._get_log(key)
            self._cleanup(log, now)
            if len(log) < self._max_requests:
                log.append(now)
                return True
            return False

    async def time_until_available(self, key: str) -> float:
        async with self._lock:
            now = time.monotonic()
            log = self._get_log(key)
            self._cleanup(log, now)
            if len(log) < self._max_requests:
                return 0.0
            oldest = log[0]
            return (oldest + self._window_seconds) - now

    async def reset(self, key: str) -> None:
        async with self._lock:
            self._logs.pop(key, None)
Compare both algorithms side by side

Both algorithms enforce rate limits, but they behave very differently under bursty traffic. The token bucket refills gradually, so a short wait can earn back a request slot. The sliding window is strict. Once you hit the limit, you wait until the oldest request falls outside the window.

  • Create a temporary file called compare.py in your project root to run both limiters and observe the difference:
import asyncio
from src.token_bucket import TokenBucketLimiter
from src.sliding_window import SlidingWindowLogLimiter


async def main():
    # Token bucket: refills 2 tokens/sec, max burst of 3
    tb = TokenBucketLimiter(rate=2.0, capacity=3)
    # Sliding window: max 3 requests per 1-second window
    sw = SlidingWindowLogLimiter(max_requests=3, window_seconds=1.0)

    print("--- Burst of 5 requests (instant) ---")
    print("         Token Bucket | Sliding Window")
    for i in range(5):
        tb_ok = await tb.is_allowed("user")
        sw_ok = await sw.is_allowed("user")
        print(f"  Req {i+1}:  {str(tb_ok):>5}        | {sw_ok}")

    print("\n--- Wait 0.5s, then try again ---")
    await asyncio.sleep(0.5)
    tb_ok = await tb.is_allowed("user")
    sw_ok = await sw.is_allowed("user")
    print(f"  After wait: {str(tb_ok):>5}    | {sw_ok}")


asyncio.run(main())

What should I see?

Both algorithms allow the first 3 requests and block requests 4 and 5. After waiting 0.5 seconds, the token bucket allows another request (it refilled 1 token at rate=2.0). The sliding window still blocks because its 1-second window hasn't expired for the oldest request yet.

This demonstrates the core tradeoff. Token bucket is forgiving of bursts and recovers quickly. Sliding window is strict and predictable.

  • Run the comparison script:
uv run python compare.py
  • You should see requests 1-3 allowed by both, requests 4-5 blocked by both, then the token bucket recovering after the wait while the sliding window stays blocked.
  • Delete compare.py when you're done. It was just for demonstration.

Seeing a ModuleNotFoundError?

Make sure you're running from the project root (the rate-limiter directory) where pyproject.toml lives. The src imports only resolve from that location.

Help me debug a ModuleNotFoundError when running my compare.py script.

Why two algorithms behind one Protocol?

This is the Strategy pattern in action. Both TokenBucketLimiter and SlidingWindowLogLimiter satisfy the same RateLimiter Protocol. Any code that depends on the Protocol can swap between algorithms without changing a single line.

In a real system, you'd pick token bucket for APIs that tolerate short bursts (like a search endpoint) and sliding window for APIs that need strict enforcement (like a payment endpoint).

You now have two production-quality rate limiting algorithms behind a clean Protocol interface. Next up, you'll integrate these into a real FastAPI application with middleware that returns proper 429 responses.

Build the FastAPI Middleware

You have two working rate limiting algorithms that can track and throttle requests by key. The missing piece is connecting them to real HTTP traffic.

In this step, you'll build a FastAPI middleware class that intercepts every incoming request, checks your limiter, and returns a proper HTTP 429 response with a Retry-After header when a client exceeds their limit.

In this step, get ready to:
  • Create the rate limiting middleware class.
  • Wire up a FastAPI app with sample endpoints and the middleware.
  • Test rate limiting by running the server and sending requests.
Create the rate limiting middleware class

Middleware sits between the client and your route handlers. Every request passes through it before reaching an endpoint, making it the ideal place to enforce rate limits.

  • Create a new file called src/middleware.py in VS Code.
  • Add the following code to define the RateLimiterMiddleware class:
from collections.abc import Callable, Awaitable

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse

from src.protocol import RateLimiter


class RateLimiterMiddleware:
    def __init__(self, app: FastAPI, limiter: RateLimiter) -> None:
        self.app = app
        self.limiter = limiter

    async def __call__(
        self,
        request: Request,
        call_next: Callable[[Request], Awaitable[Response]],
    ) -> Response:
        # Extract the client IP address from the incoming request
        client_ip = request.client.host if request.client else "unknown"
        # Check if this client is allowed to make a request
        if await self.limiter.is_allowed(client_ip):
            return await call_next(request)
        # Client exceeded their limit - calculate wait time
        retry_after = await self.limiter.time_until_available(client_ip)
        return JSONResponse(
            status_code=429,
            content={"detail": "Rate limit exceeded"},
            headers={"Retry-After": str(int(retry_after) + 1)},
        )

What does this code do?

The RateLimiterMiddleware class accepts any limiter that satisfies your RateLimiter Protocol. This means you can swap between Token Bucket and Sliding Window without changing the middleware code.

  • request.client.host extracts the client's IP address, which becomes the per-client key for rate limiting.
  • If is_allowed returns True, the request passes through to the endpoint via call_next.
  • If the client is over their limit, the middleware returns a 429 status with a Retry-After header telling the client how many seconds to wait.
  • Save src/middleware.py.

Why use the client IP as the key?

Each unique IP gets its own bucket of tokens (or its own request log). The limiter's internal dictionary maps each key to separate state, so one client hitting their limit does not affect another client's allowance.

Create the FastAPI app with endpoints

Now you need an actual API server that uses your middleware. You'll create a FastAPI app with a few endpoints and wire the rate limiter into the request lifecycle.

  • Create a new file called src/app.py in VS Code.
  • Add the following code to set up the app, limiter, middleware, and endpoints:
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse

from src.token_bucket import TokenBucketLimiter
from src.middleware import RateLimiterMiddleware

# Create the FastAPI application
app = FastAPI(title="Rate Limited API")

# Create a token bucket limiter: 2 tokens/second refill, 5 max burst
limiter = TokenBucketLimiter(rate=2.0, capacity=5)


@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    # Wrap each request through the rate limiter middleware
    middleware = RateLimiterMiddleware(app, limiter)
    return await middleware(request, call_next)


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


@app.get("/api/resource")
async def get_resource():
    return {"data": "This is a protected resource", "message": "Success"}


@app.get("/api/heavy")
async def heavy_operation():
    return {"data": "This was an expensive operation", "message": "Done"}

What does this code do?

  • TokenBucketLimiter(rate=2.0, capacity=5) creates a limiter that allows a burst of 5 requests, then refills at 2 tokens per second.
  • @app.middleware("http") registers a function that intercepts every HTTP request before it reaches a route handler.
  • The /health endpoint is a standard health check. The /api/resource and /api/heavy endpoints simulate protected resources.
  • All three endpoints share the same limiter instance, so a single client's rate limit applies across all routes.
  • Save src/app.py.

Seeing an import error?

Make sure src/__init__.py exists (it can be empty). Without it, Python won't treat src/ as a package and the from src.middleware import ... line will fail.

help me fix my import error in src/app.py

Run the server and test rate limiting

Time to see your rate limiter in action. You'll start the server and fire requests at it until you hit the limit.

  • Start the development server by running this command:
uv run uvicorn src.app:app --reload

You should see output indicating the server is running at http://127.0.0.1:8000.

  • Open a second terminal window and send a request to the health endpoint by running:
curl http://127.0.0.1:8000/health

You should see {"status":"healthy"} in the response.

  • Now exhaust the rate limit by sending 6 rapid requests to the resource endpoint:
for i in $(seq 1 6); do curl -s -o /dev/null -w "%{http_code}\n" http://127.0.0.1:8000/api/resource; done

You should see five 200 responses followed by one 429. The first 5 requests consume all tokens in the bucket, and the 6th request is rejected.

Why 5 successes then a 429?

Your limiter has capacity=5, so a fresh bucket starts with 5 tokens. Each successful request consumes one token. When the bucket is empty, the middleware returns a 429 with a Retry-After header.

If you wait a moment and try again, some tokens will have refilled at the rate of 2 per second. This is the bursty-but-fair behavior that makes Token Bucket popular in production APIs like AWS and Stripe.

  • Stop the server by pressing Ctrl+C in the terminal where uvicorn is running.

Seeing 'Address already in use'?

If port 8000 is already taken by another process, stop the other process first or specify a different port with --port 8001 at the end of the uvicorn command.

help me fix the address already in use error when running uvicorn

✔️ Awesome, I've got everything!

Great. Double check you've saved both src/middleware.py and src/app.py.

ⓧ I'd like to double check the full code

Here is the complete src/middleware.py:

from collections.abc import Callable, Awaitable

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse

from src.protocol import RateLimiter


class RateLimiterMiddleware:
    def __init__(self, app: FastAPI, limiter: RateLimiter) -> None:
        self.app = app
        self.limiter = limiter

    async def __call__(
        self,
        request: Request,
        call_next: Callable[[Request], Awaitable[Response]],
    ) -> Response:
        client_ip = request.client.host if request.client else "unknown"
        if await self.limiter.is_allowed(client_ip):
            return await call_next(request)
        retry_after = await self.limiter.time_until_available(client_ip)
        return JSONResponse(
            status_code=429,
            content={"detail": "Rate limit exceeded"},
            headers={"Retry-After": str(int(retry_after) + 1)},
        )

Here is the complete src/app.py:

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse

from src.token_bucket import TokenBucketLimiter
from src.middleware import RateLimiterMiddleware

app = FastAPI(title="Rate Limited API")

limiter = TokenBucketLimiter(rate=2.0, capacity=5)


@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    middleware = RateLimiterMiddleware(app, limiter)
    return await middleware(request, call_next)


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


@app.get("/api/resource")
async def get_resource():
    return {"data": "This is a protected resource", "message": "Success"}


@app.get("/api/heavy")
async def heavy_operation():
    return {"data": "This was an expensive operation", "message": "Done"}

Your rate limiter is now a real HTTP gatekeeper. Next up, you'll write a pytest test suite that proves your system handles concurrent load correctly.

Test with pytest and Concurrent Load

Your rate limiter and FastAPI app are built. Now you need to prove they actually work under pressure.

In this step, you'll write an async test suite that validates each algorithm individually, then hammers the API with simultaneous requests to confirm that exactly the right number get through.

In this step, get ready to:
  • Set up async API tests using httpx and pytest-asyncio.
  • Write unit tests for both rate limiting algorithms.
  • Fire concurrent requests with asyncio.TaskGroup and assert the rate limiter holds.
Set up the async test client with httpx

Your pyproject.toml already has pytest-asyncio configured with asyncio_mode = "auto". This means any async def test_ function is automatically treated as an asyncio test. No decorators needed.

What does auto mode do?

In auto mode, pytest-asyncio automatically adds the asyncio marker to all asynchronous test functions. You can write plain async def test_something() functions and they just work.

The asyncio_default_fixture_loop_scope = "function" setting gives each test its own event loop for isolation.

To test the FastAPI app without starting a real server, you'll use httpx.AsyncClient with httpx.ASGITransport. This sends requests directly to the ASGI app in-process.

  • Create a new file called tests/test_api.py and add the following code:
import asyncio

import httpx
from fastapi.testclient import TestClient

from src.app import app, limiter


async def test_health_returns_200():
    # Create an async client that talks directly to the ASGI app
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test"
    ) as client:
        # Reset the limiter so previous test runs don't interfere
        await limiter.reset("127.0.0.1")
        response = await client.get("/health")
        assert response.status_code == 200
        assert response.json() == {"status": "healthy"}


async def test_rate_limit_returns_429():
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test"
    ) as client:
        await limiter.reset("127.0.0.1")
        # Exhaust all 5 tokens in the bucket
        for _ in range(5):
            await client.get("/api/resource")
        # The 6th request should be rate limited
        response = await client.get("/api/resource")
        assert response.status_code == 429
        assert "Retry-After" in response.headers
        assert response.json() == {"detail": "Rate limit exceeded"}

What does this code do?

httpx.ASGITransport(app=app) lets httpx send HTTP requests directly to your FastAPI app without a network socket. The test client behaves exactly like a real HTTP client, but everything stays in-process.

Each test calls await limiter.reset("127.0.0.1") first. This clears any tokens consumed by a previous test so each test starts with a fresh bucket.

  • Save tests/test_api.py.
  • Run these two tests by running:
uv run pytest tests/test_api.py -v
  • You should see both tests passing with green output.

Seeing import errors?

Make sure you're running uv run pytest from the rate-limiter project root (the folder containing pyproject.toml). If pytest can't find src.app, check that your src/__init__.py file exists.

help me debug pytest import errors for my rate limiter project

Write tests for the token bucket and sliding window algorithms

Unit tests for each algorithm verify the core behaviors in isolation, without the middleware or HTTP layer involved.

  • Create a new file called tests/test_token_bucket.py and add the following code:
import asyncio

from src.token_bucket import TokenBucketLimiter


async def test_allows_requests_up_to_capacity():
    # A bucket with capacity 3 should allow 3 requests immediately
    limiter = TokenBucketLimiter(rate=1.0, capacity=3)
    results = [await limiter.is_allowed("user1") for _ in range(3)]
    assert all(results)


async def test_blocks_after_capacity_exhausted():
    # After 3 requests, the 4th should be blocked
    limiter = TokenBucketLimiter(rate=1.0, capacity=3)
    for _ in range(3):
        await limiter.is_allowed("user1")
    assert await limiter.is_allowed("user1") is False


async def test_refills_over_time():
    # With rate=10 tokens/sec, waiting 0.15s should refill at least 1 token
    limiter = TokenBucketLimiter(rate=10.0, capacity=3)
    for _ in range(3):
        await limiter.is_allowed("user1")
    assert await limiter.is_allowed("user1") is False
    await asyncio.sleep(0.15)
    assert await limiter.is_allowed("user1") is True


async def test_different_keys_are_independent():
    # Exhausting user1's bucket should not affect user2
    limiter = TokenBucketLimiter(rate=1.0, capacity=2)
    for _ in range(2):
        await limiter.is_allowed("user1")
    assert await limiter.is_allowed("user1") is False
    assert await limiter.is_allowed("user2") is True


async def test_reset_restores_capacity():
    # Resetting a key gives it a fresh bucket
    limiter = TokenBucketLimiter(rate=1.0, capacity=3)
    for _ in range(3):
        await limiter.is_allowed("user1")
    await limiter.reset("user1")
    assert await limiter.is_allowed("user1") is True

What does this code do?

These five tests cover the core token bucket behaviors: burst allowance up to capacity, blocking when exhausted, refill over time, key isolation between different clients, and reset functionality.

The test_refills_over_time test uses a high rate (10 tokens/sec) and a short sleep (0.15s) so the test finishes quickly while still verifying time-based refill.

  • Save tests/test_token_bucket.py.

Now write the equivalent tests for the sliding window algorithm.

  • Create a new file called tests/test_sliding_window.py and add the following code:
import asyncio

from src.sliding_window import SlidingWindowLogLimiter


async def test_allows_up_to_max_requests():
    # Should allow exactly max_requests within the window
    limiter = SlidingWindowLogLimiter(max_requests=3, window_seconds=1.0)
    results = [await limiter.is_allowed("user1") for _ in range(3)]
    assert all(results)


async def test_blocks_after_max_requests():
    # The 4th request within the window should be blocked
    limiter = SlidingWindowLogLimiter(max_requests=3, window_seconds=1.0)
    for _ in range(3):
        await limiter.is_allowed("user1")
    assert await limiter.is_allowed("user1") is False


async def test_allows_after_window_expires():
    # After the window passes, old requests fall off and new ones are allowed
    limiter = SlidingWindowLogLimiter(max_requests=2, window_seconds=0.1)
    for _ in range(2):
        await limiter.is_allowed("user1")
    assert await limiter.is_allowed("user1") is False
    await asyncio.sleep(0.15)
    assert await limiter.is_allowed("user1") is True


async def test_time_until_available_returns_wait():
    # Should return a positive wait time when limit is hit
    limiter = SlidingWindowLogLimiter(max_requests=1, window_seconds=1.0)
    await limiter.is_allowed("user1")
    wait = await limiter.time_until_available("user1")
    assert 0.0 < wait <= 1.0

What does this code do?

These tests mirror the token bucket tests but validate sliding window behavior. The key difference is test_time_until_available_returns_wait which checks that the limiter correctly reports how long a client must wait before their next request will be allowed.

Short window values (0.1s) keep the tests fast while still proving the window-expiry logic works.

  • Save tests/test_sliding_window.py.
  • Run both algorithm test files by running:
uv run pytest tests/test_token_bucket.py tests/test_sliding_window.py -v
  • You should see all 9 tests passing (5 token bucket, 4 sliding window).

Tests failing with timing errors?

If test_refills_over_time or test_allows_after_window_expires fails intermittently, your machine might be under heavy load. Try running the tests again. The sleep values are deliberately generous (0.15s for refills that only need 0.1s) to avoid flakiness.

help me debug flaky async timing tests in my rate limiter

Write a concurrency test with asyncio.TaskGroup

The real test of a rate limiter is concurrent access. A single-threaded loop test might pass, but real production traffic arrives simultaneously. You need to prove that even under concurrent load, exactly the right number of requests get through.

  • In tests/test_api.py, find the last line of test_rate_limit_returns_429:
        assert response.json() == {"detail": "Rate limit exceeded"}
  • Add the following test function below it (leave two blank lines between the functions):
async def test_concurrent_requests_respect_limit():
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test"
    ) as client:
        await limiter.reset("127.0.0.1")
        # Fire 10 requests simultaneously using TaskGroup
        async with asyncio.TaskGroup() as tg:
            tasks = [
                tg.create_task(client.get("/api/resource")) for _ in range(10)
            ]
        # Collect results from all completed tasks
        responses = [t.result() for t in tasks]
        successes = [r for r in responses if r.status_code == 200]
        rate_limited = [r for r in responses if r.status_code == 429]
        # With capacity=5, exactly 5 should succeed and 5 should get 429
        assert len(successes) == 5
        assert len(rate_limited) == 5

What does this code do?

asyncio.TaskGroup fires all 10 HTTP requests at the same moment. Because the token bucket's asyncio.Lock serializes access, exactly 5 requests consume the 5 available tokens. The remaining 5 requests find the bucket empty and receive 429 responses.

This is the test that proves your asyncio.Lock is working correctly. Without the lock, you'd see race conditions where more than 5 requests slip through.

  • Save tests/test_api.py.
  • Run the entire test suite by running:
uv run pytest -v
  • You should see all 12 tests passing (5 token bucket, 4 sliding window, 3 API tests).

Concurrent test asserting wrong numbers?

If you see a different split than 5 successes and 5 rate-limited, check that your limiter.reset("127.0.0.1") call is present. Without it, leftover state from the previous test (which exhausted all tokens) carries over.

help me debug my concurrent rate limiter test that's asserting wrong numbers

✔️ Awesome, I've got everything!

Great. Double check that you've saved all three test files.

ⓧ I'd like to double check the full code

Here is the full tests/test_api.py:

import asyncio

import httpx
from fastapi.testclient import TestClient

from src.app import app, limiter


async def test_health_returns_200():
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test"
    ) as client:
        await limiter.reset("127.0.0.1")
        response = await client.get("/health")
        assert response.status_code == 200
        assert response.json() == {"status": "healthy"}


async def test_rate_limit_returns_429():
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test"
    ) as client:
        await limiter.reset("127.0.0.1")
        for _ in range(5):
            await client.get("/api/resource")
        response = await client.get("/api/resource")
        assert response.status_code == 429
        assert "Retry-After" in response.headers
        assert response.json() == {"detail": "Rate limit exceeded"}


async def test_concurrent_requests_respect_limit():
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test"
    ) as client:
        await limiter.reset("127.0.0.1")
        async with asyncio.TaskGroup() as tg:
            tasks = [
                tg.create_task(client.get("/api/resource")) for _ in range(10)
            ]
        responses = [t.result() for t in tasks]
        successes = [r for r in responses if r.status_code == 200]
        rate_limited = [r for r in responses if r.status_code == 429]
        assert len(successes) == 5
        assert len(rate_limited) == 5

Here is the full tests/test_token_bucket.py:

import asyncio

from src.token_bucket import TokenBucketLimiter


async def test_allows_requests_up_to_capacity():
    limiter = TokenBucketLimiter(rate=1.0, capacity=3)
    results = [await limiter.is_allowed("user1") for _ in range(3)]
    assert all(results)


async def test_blocks_after_capacity_exhausted():
    limiter = TokenBucketLimiter(rate=1.0, capacity=3)
    for _ in range(3):
        await limiter.is_allowed("user1")
    assert await limiter.is_allowed("user1") is False


async def test_refills_over_time():
    limiter = TokenBucketLimiter(rate=10.0, capacity=3)
    for _ in range(3):
        await limiter.is_allowed("user1")
    assert await limiter.is_allowed("user1") is False
    await asyncio.sleep(0.15)
    assert await limiter.is_allowed("user1") is True


async def test_different_keys_are_independent():
    limiter = TokenBucketLimiter(rate=1.0, capacity=2)
    for _ in range(2):
        await limiter.is_allowed("user1")
    assert await limiter.is_allowed("user1") is False
    assert await limiter.is_allowed("user2") is True


async def test_reset_restores_capacity():
    limiter = TokenBucketLimiter(rate=1.0, capacity=3)
    for _ in range(3):
        await limiter.is_allowed("user1")
    await limiter.reset("user1")
    assert await limiter.is_allowed("user1") is True

Here is the full tests/test_sliding_window.py:

import asyncio

from src.sliding_window import SlidingWindowLogLimiter


async def test_allows_up_to_max_requests():
    limiter = SlidingWindowLogLimiter(max_requests=3, window_seconds=1.0)
    results = [await limiter.is_allowed("user1") for _ in range(3)]
    assert all(results)


async def test_blocks_after_max_requests():
    limiter = SlidingWindowLogLimiter(max_requests=3, window_seconds=1.0)
    for _ in range(3):
        await limiter.is_allowed("user1")
    assert await limiter.is_allowed("user1") is False


async def test_allows_after_window_expires():
    limiter = SlidingWindowLogLimiter(max_requests=2, window_seconds=0.1)
    for _ in range(2):
        await limiter.is_allowed("user1")
    assert await limiter.is_allowed("user1") is False
    await asyncio.sleep(0.15)
    assert await limiter.is_allowed("user1") is True


async def test_time_until_available_returns_wait():
    limiter = SlidingWindowLogLimiter(max_requests=1, window_seconds=1.0)
    await limiter.is_allowed("user1")
    wait = await limiter.time_until_available("user1")
    assert 0.0 < wait <= 1.0

All 12 tests green. You've proven that your rate limiter is correct under isolation and under concurrent load. Next up, you'll take on a bonus challenge that brings per-endpoint granularity to your rate limiting.

Secret mission

Add a Per-Endpoint Decorator Rate Limiter

Your global rate limiter treats every endpoint the same. But in production, a search endpoint can handle far more traffic than a create endpoint. Can you build a decorator that lets you set different limits on each route?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your project files to continue experimenting, or delete them to free up disk space.

Resources you used:

  • The rate-limiter project directory (source code, tests, virtual environment).

Keep everything running

No action needed. Choose this if you want to keep experimenting with your rate limiter, add new algorithms, or extend the FastAPI app.

  • Your project files stay in the rate-limiter folder on your Desktop.
  • The virtual environment managed by uv stays inside the project directory.

Pause - I'll come back to this later

Since there are no running processes or cloud resources, pausing is the same as keeping. Your files sit idle on disk until you return.

  • If you have a uvicorn dev server running in your terminal, stop it with Ctrl+C.
  • Everything else stays as-is. No ongoing costs.

Delete - I don't want to use this again

Remove the entire project directory to start fresh. This deletes all source code, tests, and the virtual environment.

  • Stop any running server with Ctrl+C in the terminal.
  • Delete the project folder by running this command from your Desktop:

macOS

cd ~/Desktop
rm -rf rate-limiter

Windows

cd ~\Desktop
Remove-Item -Recurse -Force rate-limiter

Nice Work!

Nice Work!

You just built exactly what interviewers ask about in system design rounds. You did it with production-quality Python that demonstrates async mastery, clean architecture, and thorough testing.

You've learned how to:

  • Build two rate limiting algorithms (Token Bucket and Sliding Window Log) behind a clean Protocol interface, making them swappable via the Strategy pattern.
  • Integrate a FastAPI middleware that enforces per-client rate limits and returns standards-compliant 429 responses with Retry-After headers.
  • Test concurrent access with asyncio.TaskGroup and httpx.AsyncClient, proving your limiter is safe under parallel load.
  • Added per-endpoint rate limiting with a custom @rate_limit decorator that applies different limits to different routes.

Ready to quiz yourself?