FastAPI has quickly become the go‑to framework for building high‑performance Python APIs, thanks to its async‑first design, automatic OpenAPI documentation, and developer‑friendly syntax. However, as your service scales, protecting it from abuse, accidental overload, or malicious traffic becomes essential. That’s where rate limiting comes in – a technique that caps the number of requests a client can make within a defined time window. In this tutorial we’ll walk through everything you need to know to implement robust rate limiting in a FastAPI application, from simple in‑memory counters to production‑ready Redis‑backed solutions.
Why Rate Limiting Matters for FastAPI APIs
- Prevent resource exhaustion: Limit the impact of spikes that could otherwise exhaust CPU, memory, or database connections.
- Guard against abuse: Thwart brute‑force attacks, credential stuffing, and denial‑of‑service attempts.
- Ensure fair usage: Give every client a predictable level of service, especially in public APIs.
- Improve user experience: By returning clear HTTP 429 responses, clients know when to back off instead of experiencing timeouts.
Core Concepts Behind Rate Limiting
Fixed Window vs. Sliding Window
A fixed window resets counters at regular intervals (e.g., every minute). It’s simple but can cause bursts at the boundary. A sliding window tracks requests over the last N seconds, offering smoother throttling.
Token Bucket Algorithm
The token bucket algorithm combines burst flexibility with a steady refill rate. Each request consumes a token; tokens are added back at a fixed rate, up to a maximum capacity. This is ideal for APIs that need occasional spikes without breaking limits.
Choosing the Right Storage Backend
The storage layer determines how your limits behave in a distributed environment. Here are three common options:
- In‑memory dictionary: Fast and easy for single‑process development, but not shared across workers.
- SQLite or file‑based store: Slightly more persistent, still limited to a single machine.
- Redis: The de‑facto standard for distributed rate limiting – fast, atomic operations, and supports expiration.
Implementing a Simple In‑Memory Rate Limiter
Let’s start with a minimal example that works for local testing. We’ll use a dict to store request counts per IP address.
from fastapi import FastAPI, Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
import time
app = FastAPI()
# Settings
MAX_REQUESTS = 10 # allowed requests
WINDOW_SECONDS = 60 # per minute
# In‑memory store: {client_ip: (first_request_timestamp, count)}
request_counts = {}
class SimpleRateLimitMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
client_ip = request.client.host
now = time.time()
# Initialize or reset the window
if client_ip not in request_counts or now - request_counts[client_ip][0] > WINDOW_SECONDS:
request_counts[client_ip] = (now, 1)
else:
_, count = request_counts[client_ip]
request_counts[client_ip] = (request_counts[client_ip][0], count + 1)
# Enforce limit
if request_counts[client_ip][1] > MAX_REQUESTS:
raise HTTPException(
status_code=429,
detail=f"Rate limit exceeded. Try again in {int(WINDOW_SECONDS - (now - request_counts[client_ip][0]))} seconds."
)
response = await call_next(request)
return response
app.add_middleware(SimpleRateLimitMiddleware)
@app.get("/hello")
async def hello():
return {"message": "Hello, FastAPI!"}
This middleware checks the client’s IP, updates the request count, and raises a 429 Too Many Requests error when the limit is breached. While great for demos, it won’t work when you scale to multiple workers or containers because each process maintains its own dictionary.
Building a Production‑Ready Redis Rate Limiter
Redis provides atomic increment and expiration commands that make implementing the token bucket algorithm straightforward.
Prerequisites
- Python 3.9+ installed.
- FastAPI and
uvicornfor serving. redisPython client (e.g.,redis-py).- A running Redis instance (Docker is the easiest way).
Docker Quick‑Start for Redis
docker run -d --name fastapi-redis -p 6379:6379 redis:7-alpine
Redis Middleware Code
Below is a reusable middleware class that you can drop into any FastAPI project.
import aioredis
from fastapi import FastAPI, Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
import os
app = FastAPI()
# Environment variables for flexibility
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
MAX_TOKENS = int(os.getenv("MAX_TOKENS", 100)) # burst capacity
REFILL_RATE = float(os.getenv("REFILL_RATE", 1.0)) # tokens per second
class RedisRateLimitMiddleware(BaseHTTPMiddleware):
def __init__(self, app):
super().__init__(app)
self.redis = None
async def dispatch(self, request: Request, call_next):
if self.redis is None:
self.redis = await aioredis.from_url(REDIS_URL, decode_responses=True)
client_id = request.client.host
key = f"rl:{client_id}"
now = int(time.time() * 1000) # milliseconds
# Lua script ensures atomic check‑and‑update
lua_script = """
local key = KEYS[1]
local max_tokens = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local data = redis.call('HMGET', key, 'tokens', 'timestamp')
local tokens = tonumber(data[1])
local timestamp = tonumber(data[2])
if tokens == nil then
tokens = max_tokens
timestamp = now
else
local elapsed = now - timestamp
local refill = (elapsed / 1000) * refill_rate
tokens = math.min(max_tokens, tokens + refill)
timestamp = now
end
if tokens < 1 then
redis.call('HMSET', key, 'tokens', tokens, 'timestamp', timestamp)
redis.call('PEXPIRE', key, 60000) -- keep key for 60s
return -1
else
tokens = tokens - 1
redis.call('HMSET', key, 'tokens', tokens, 'timestamp', timestamp)
redis.call('PEXPIRE', key, 60000)
return tokens
end
"""
remaining = await self.redis.eval(lua_script, 1, key, MAX_TOKENS, REFILL_RATE, now)
if remaining < 0:
raise HTTPException(
status_code=429,
detail="Rate limit exceeded. Please slow down."
)
response = await call_next(request)
# Optional: expose remaining tokens in a header
response.headers["X-RateLimit-Remaining"] = str(int(remaining))
return response
app.add_middleware(RedisRateLimitMiddleware)
@app.get("/items")
async def list_items():
return {"items": ["apple", "banana", "cherry"]}
The Lua script runs entirely inside Redis, guaranteeing that token calculation and decrement are atomic even under heavy concurrency. The middleware also adds an X-RateLimit-Remaining header so clients can programmatically adjust their request pacing.
Customizing Limits Per Endpoint or User
Often you’ll need different limits for public vs. internal endpoints, or per‑API‑key. FastAPI’s dependency injection makes this simple.
from fastapi import Depends
def get_rate_limit(request: Request):
# Example: higher limit for admin IPs
admin_ips = {"127.0.0.1"}
if request.client.host in admin_ips:
return {"max_tokens": 1000, "refill_rate": 20}
return {"max_tokens": 100, "refill_rate": 1}
class DynamicRedisRateLimitMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
limit = await get_rate_limit(request)
# Pass limit values to the Lua script (similar to previous example)
# ... (omitted for brevity)
return await super().dispatch(request, call_next)
By injecting a dependency that inspects request headers, query parameters, or authentication tokens, you can tailor the rate‑limit policy on a per‑user basis.
Testing Your Rate Limiter
Automated tests help you ensure the limiter behaves as expected. Below is a pytest example using httpx and asyncio.
import pytest
import httpx
import asyncio
@pytest.mark.asyncio
async def test_rate_limit_exceeded():
async with httpx.AsyncClient(app=app, base_url="http://test") as client:
# Assuming limit is 5 requests per minute
for _ in range(5):
resp = await client.get("/items")
assert resp.status_code == 200
# 6th request should be blocked
resp = await client.get("/items")
assert resp.status_code == 429
assert resp.json()["detail"] == "Rate limit exceeded. Please slow down."
Run the test with pytest -s. Adjust the loop count to match your configured limits.
Leave a Reply