When you start building production‑ready services with Python FastAPI, one of the first concerns is protecting your endpoints from abuse. Whether it’s a malicious bot hammering your API or an honest user accidentally spamming requests, uncontrolled traffic can quickly degrade performance, increase costs, and even cause downtime. This is where rate limiting comes into play. In this tutorial we’ll walk through everything you need to know to implement robust API rate limiting in FastAPI – from the theory behind it, to choosing the right storage backend, and finally wiring up a reusable dependency that you can drop into any route.
Why Rate Limiting Matters for FastAPI Applications
FastAPI is renowned for its speed, thanks to Starlette and ASGI, but speed alone doesn’t guarantee stability. Here are three core reasons why you should add rate limiting to every public FastAPI endpoint:
- Prevent resource exhaustion: Limit the number of requests per second or minute to keep CPU, memory, and database connections within safe bounds.
- Guard against denial‑of‑service attacks: Simple throttling can stop many brute‑force and scraping attacks before they reach your business logic.
- Enforce fair usage policies: If you offer tiered plans (free vs. premium), rate limits let you differentiate service levels without custom code per user.
Core Concepts Behind Rate Limiting
1. Tokens, Buckets, and Windows
Most rate‑limiting algorithms fall into two families:
- Fixed window: Count requests in a static time slice (e.g., 100 requests per minute). Easy to implement but can cause “burst” spikes at the edge of windows.
- Sliding window / token bucket: Distribute tokens over time, allowing smooth bursts while still respecting an overall average rate.
For most FastAPI projects, the token bucket approach offers the best balance of fairness and simplicity, especially when combined with a fast in‑memory store like Redis.
2. Choosing a Storage Backend
Rate‑limit counters must be shared across all workers and processes. Common choices include:
- In‑memory (process local): Good for development or single‑process deployments, but not safe for scaled environments.
- Redis: Fast, supports atomic increment operations, and works seamlessly with distributed FastAPI deployments.
- Memcached: Similar to Redis but lacks built‑in Lua scripting for complex logic.
- Database tables: Viable for low‑traffic APIs, but adds latency and can lock rows under heavy load.
In this tutorial we’ll use redis-py because it’s widely adopted, performant, and integrates nicely with async FastAPI code.
Step‑by‑Step Implementation
Prerequisites
- Python 3.9+ installed.
- FastAPI and Uvicorn:
pip install fastapi uvicorn - Redis server running locally or remotely.
- Redis client library:
pip install redis[async]
1. Setting Up a Redis Connection
# redis_client.py
import os
import redis.asyncio as redis
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
async def get_redis() -> redis.Redis:
return redis.from_url(REDIS_URL, decode_responses=True)
This async helper returns a shared Redis instance that can be injected into FastAPI dependencies.
2. Creating a Rate‑Limit Dependency
FastAPI’s dependency injection system makes it trivial to run logic before a request reaches the path operation. Below is a reusable rate_limit dependency that implements a token‑bucket algorithm using Redis Lua scripting for atomicity.
# rate_limiter.py
import time
from fastapi import HTTPException, Request, Depends
from redis.asyncio import Redis
from .redis_client import get_redis
# Lua script that:
# 1. Retrieves the current token count.
# 2. Adds tokens based on elapsed time.
# 3. Checks if a token can be consumed.
# 4. Stores the new state.
RATE_LIMIT_SCRIPT = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local period = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local bucket = redis.call('HMGET', key, 'tokens', 'timestamp')
local tokens = tonumber(bucket[1]) or limit
local timestamp = tonumber(bucket[2]) or now
local elapsed = now - timestamp
local refill = (elapsed / period) * limit
tokens = math.min(tokens + refill, limit)
if tokens < 1 then
return -1
else
tokens = tokens - 1
redis.call('HMSET', key, 'tokens', tokens, 'timestamp', now)
redis.call('EXPIRE', key, period)
return tokens
end
"""
async def rate_limit(
request: Request,
redis: Redis = Depends(get_redis),
limit: int = 100, # max requests
period: int = 60 # per seconds
):
"""
Enforces a token‑bucket rate limit.
"""
client_ip = request.client.host
key = f"rl:{client_ip}"
now = int(time.time())
# Execute Lua script atomically
remaining = await redis.eval(
RATE_LIMIT_SCRIPT,
1,
key,
limit,
period,
now,
)
if remaining < 0:
raise HTTPException(
status_code=429,
detail="Too Many Requests – rate limit exceeded",
headers={"Retry-After": str(period)},
)
# Optionally expose remaining tokens in response headers
request.state.rate_limit_remaining = remaining
Key points:
- The script stores two fields per IP:
tokensandtimestamp. - Redis
EXPIREautomatically cleans up stale keys after the period. - We raise
HTTPException(429)when the bucket is empty.
3. Applying the Dependency to Endpoints
# main.py
from fastapi import FastAPI, Depends, Response
from .rate_limiter import rate_limit
app = FastAPI(title="FastAPI Rate Limiting Demo")
@app.get("/items")
async def list_items(response: Response, _: None = Depends(rate_limit)):
# Access remaining tokens from request.state if you want to expose it
remaining = response.request.state.rate_limit_remaining
response.headers["X-RateLimit-Remaining"] = str(remaining)
return {"message": "Here are your items", "rate_limit_remaining": remaining}
Every request to /items now passes through rate_limit. If the client exceeds 100 requests per minute, FastAPI automatically returns a 429 Too Many Requests response.
4. Customizing Limits Per Route or User
Because rate_limit accepts limit and period as parameters, you can create route‑specific or user‑specific policies:
@app.get("/premium-data")
async def premium_data(_: None = Depends(lambda: rate_limit(limit=1000, period=60))):
return {"data": "Premium content available to paid users"}
For authenticated users you could key the Redis bucket by user_id instead of IP, providing per‑account quotas.
Testing Your Rate Limiter
Manual Curl Test
Run the server with uvicorn main:app --reload and execute:
for i in {1..105}; do
curl -s -o /dev/null -w "%{http_code} " http://localhost:8000/items
done
echo
You should see a series of 200 status codes followed by a few 429 responses once the limit is hit.
Automated pytest Example
# test_rate_limit.py
import pytest
from httpx import AsyncClient
from main import app
@pytest.mark.asyncio
async def test_rate_limit():
async with AsyncClient(app=app, base_url="http://test") as client:
for i in range(101):
resp = await client.get("/items")
if i < 100:
assert resp.status_code == 200
else:
assert resp.status_code == 429
This test confirms that the 101st request is blocked, proving the limiter works as expected.
Advanced Tips & Best Practices
- Use a CDN or API gateway: Services like Cloudflare, AWS API Gateway, or Kong can enforce rate limits at the edge, reducing load on your FastAPI workers.
- Separate limits for read vs. write: Apply stricter limits on POST/PUT routes that modify data, while allowing higher read throughput.
- Grace periods for authenticated users: Store a “burst” token count for premium accounts to allow occasional spikes.
- Monitoring and alerts: Export Redis key metrics (e.g., hit rate, evictions) to Prometheus and set alerts when 429 rates climb suddenly.
- Cache the Lua script SHA: Loading the script once and re‑using its SHA hash reduces network round‑trips.
Deploying to Production
When you move beyond local development, keep these deployment considerations in
Leave a Reply