Python Fastapi Api Rate Limiting Tutorial

If you’re building a high‑traffic REST service with FastAPI, you’ll quickly discover that protecting your endpoints from abuse is just as important as delivering blazing‑fast responses. Rate limiting—the practice of restricting how many requests a client can make within a given time window—keeps your API reliable, prevents denial‑of‑service attacks, and ensures fair usage for all consumers. In this tutorial we’ll walk through everything you need to know to add robust, production‑ready rate limiting to a FastAPI application, from the basics of why it matters to hands‑on code using the popular slowapi library and a custom middleware solution.

What Is Rate Limiting and Why It Matters

Rate limiting is a defensive technique that controls the number of requests a client (identified by IP address, API key, or user token) can send to an endpoint over a defined period. Here are the primary reasons every API should enforce it:

• Protect resources: Prevent a single client from overwhelming your database, cache, or third‑party services.
• Improve user experience: By throttling abusive traffic, legitimate users experience more consistent latency.
• Guard against attacks: Simple brute‑force attacks, credential stuffing, and DDoS attempts become harder to execute.
• Enforce business rules: Offer tiered plans (free vs. premium) with different request quotas.

FastAPI Overview: A Quick Recap

FastAPI is a modern, async‑first web framework built on top of Starlette and Pydantic. Its key strengths for API development include:

• Automatic OpenAPI documentation.
• Dependency injection that makes adding cross‑cutting concerns (like authentication or rate limiting) straightforward.
• Native support for asynchronous endpoints, which pairs nicely with high‑performance rate‑limiting backends such as Redis.

Because FastAPI is built on Starlette, any Starlette Middleware or ExceptionHandler can be reused directly, giving us flexibility when choosing a rate‑limiting strategy.

Choosing a Rate Limiting Strategy

Before diving into code, decide which strategy aligns with your use case. The most common patterns are:

1. Fixed window: Allow N requests per fixed interval (e.g., 100 requests per minute). Simple but can cause bursts at the window boundary.
2. Sliding window log: Record timestamps of each request and count those within the last interval. Accurate but memory‑intensive.
3. Token bucket / Leaky bucket: Tokens are added at a steady rate; each request consumes a token. Handles bursts gracefully.

For most web APIs, a fixed window with Redis backing offers a good balance of simplicity and performance. The slowapi library implements this pattern out of the box.

Implementing Rate Limiting with slowapi

SlowAPI is a thin wrapper around the popular limits library, providing ready‑made FastAPI decorators and middleware. Follow these steps to get it up and running.

Step 1: Install Dependencies

pip install fastapi[all] slowapi redis

Step 2: Configure Redis Backend

Redis stores the request counters. You can run it locally with Docker:

docker run -d -p 6379:6379 redis:7

Step 3: Create the FastAPI App with SlowAPI Middleware

from fastapi import FastAPI, Request, HTTPException
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

# Initialise the limiter – 100 requests per minute per IP
limiter = Limiter(key_func=get_remote_address, default_limits=["100/minute"])
app = FastAPI()

# Register the global exception handler for rate‑limit errors
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

Step 4: Protect Endpoints with Decorators

You can apply a global limit (as above) or fine‑tune per‑route limits using the @limiter.limit decorator.

@app.get("/items")
@limiter.limit("10/minute")
async def list_items():
    return {"msg": "This endpoint is limited to 10 requests per minute per IP."}

@app.post("/orders")
@limiter.limit("5/minute")
async def create_order(order: dict):
    # Business logic here
    return {"status": "order created"}

Step 5: Testing the Limits

Run the app with uvicorn main:app --reload and use curl or a tool like hey to fire requests. After exceeding the limit you’ll receive a 429 Too Many Requests response with a helpful Retry-After header.

Custom Middleware Approach (When You Need Full Control)

While slowapi covers most scenarios, some projects require custom logic—such as rate limiting based on API keys, user roles, or dynamic quotas stored in a database. Below is a lightweight middleware example that demonstrates how to implement a token‑bucket algorithm with Redis.

Middleware Code

import time
import aioredis
from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

class TokenBucketMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, redis_url: str, capacity: int = 100, refill_rate: float = 1.0):
        super().__init__(app)
        self.capacity = capacity          # Max tokens per bucket
        self.refill_rate = refill_rate    # Tokens added per second
        self.redis = aioredis.from_url(redis_url, decode_responses=True)

    async def dispatch(self, request: Request, call_next):
        identifier = request.client.host   # Could be API key or user ID instead
        bucket_key = f"rate_limit:{identifier}"

        # 1. Fetch current token count and last refill timestamp
        data = await self.redis.hmget(bucket_key, "tokens", "timestamp")
        tokens = float(data[0]) if data[0] else self.capacity
        last_ts = float(data[1]) if data[1] else time.time()

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

        # 3. Check if a token is available
        if tokens < 1:
            # Not enough tokens – reject request
            retry_after = int((1 - tokens) / self.refill_rate) + 1
            return Response(
                content='{"detail":"Rate limit exceeded"}',
                status_code=429,
                media_type="application/json",
                headers={"Retry-After": str(retry_after)},
            )

        # 4. Consume a token and persist new state
        tokens -= 1
        await self.redis.hmset_dict(bucket_key, {"tokens": tokens, "timestamp": now})
        # Optional: set TTL to automatically clean stale keys
        await self.redis.expire(bucket_key, int(self.capacity / self.refill_rate * 2))

        # 5. Continue processing the request
        response = await call_next(request)
        return response

app = FastAPI()
app.add_middleware(
    TokenBucketMiddleware,
    redis_url="redis://localhost:6379",
    capacity=200,          # 200 requests per 200 seconds ≈ 1 rps
    refill_rate=1.0        # 1 token added each second
)

@app.get("/ping")
async def ping():
    return {"msg": "pong"}

Why Use Custom Middleware?

• Dynamic quotas: Pull per‑user limits from a database and adjust capacity on the fly.
• Multi‑tenant SaaS: Enforce different limits for free vs. paid plans without adding extra decorators.
• Fine‑grained logging: Record each limit breach for analytics or billing.

Testing Your Rate Limiting Logic

Automated tests ensure that limits behave as expected under load. Below is a simple pytest example using httpx and the FastAPI test client.

import pytest
from fastapi.testclient import TestClient
from main import app   # assuming your FastAPI instance is in main.py

client = TestClient(app)

def test_fixed_limit():
    # The endpoint is limited to 5 requests per minute
    for i in range(5):
        resp = client.get("/orders")
        assert resp.status_code == 200

    # Sixth request should be blocked
    resp = client.get("/orders")
    assert resp.status_code == 429
    assert "Retry-After" in resp.headers

Run the test suite with pytest -q. For more realistic load testing, tools like hey or Locust can simulate thousands of concurrent users.

Best Practices for Production