Rate Limiting & Throttling APIs

Rate limiting protects your API from abuse, DoS attacks, and resource exhaustion. This guide covers practical implementation with Redis and middleware patterns you can use in FastAPI or Django.

1. Why Rate Limiting Matters

Without rate limiting, a single client can:

Crash your server with thousands of requests
Scrape all your data
Brute-force authentication endpoints
Cause excessive database load

Rate limiting enforces rules like "max 100 requests per minute per user" or "max 10 login attempts per 15 minutes."

Reference: Redis Official Site

2. Common Rate Limiting Strategies

2.1 Fixed Window

Count requests in a fixed time window (e.g., 1 minute). Simple but can have edge cases at window boundaries.

# Pseudocode
minute = current_time / 60
key = f"requests:{user_id}:{minute}"
requests = redis.incr(key)
if requests > LIMIT:
return 429 Too Many Requests

2.2 Sliding Window

More accurate. Track requests with timestamps and count only recent ones.

# Pseudocode
key = f"requests:{user_id}"
now = current_time()
redis.zremrangebyscore(key, 0, now - 60)  # Remove old requests
count = redis.zcard(key)
if count >= LIMIT:
return 429
redis.zadd(key, {now: count + 1})

2.3 Token Bucket

Allows bursts while maintaining average rate. Tokens refill over time.

# Pseudocode
key = f"tokens:{user_id}"
tokens = redis.get(key) or MAX_TOKENS
elapsed = current_time() - last_check
tokens = min(MAX_TOKENS, tokens + elapsed * REFILL_RATE)
if tokens < 1:
return 429
redis.decrby(key, 1)

Reference: Token Bucket Algorithm (Wikipedia)

3. FastAPI Implementation with Redis

3.1 Installation

pip install fastapi uvicorn redis slowapi

slowapi is a popular rate limiting library for FastAPI built on top of werkzeug.

3.2 Simple Rate Limiting with slowapi

from fastapi import FastAPI
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from fastapi.responses import JSONResponse
app = FastAPI()
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
@app.exception_handler(RateLimitExceeded)
async def rate_limit_handler(request, exc):
return JSONResponse(
status_code=429,
content={"detail": "Rate limit exceeded. Try again later."}
)
@app.get("/api/data")
@limiter.limit("10/minute")
async def get_data(request):
return {"data": "value"}

This limits /api/data to 10 requests per minute per IP address.

3.3 Custom Redis-Based Rate Limiter

import redis
import time
from fastapi import FastAPI, HTTPException, Request
app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379, db=0)
RATE_LIMIT = 100  # requests
WINDOW = 60  # seconds
async def check_rate_limit(identifier: str):
key = f"rate_limit:{identifier}"
current = redis_client.get(key)
if current is None:
redis_client.setex(key, WINDOW, 1)
return True
if int(current) < RATE_LIMIT:
redis_client.incr(key)
return True
return False
@app.get("/api/protected")
async def protected_endpoint(request: Request):
client_ip = request.client.host
if not await check_rate_limit(client_ip):
raise HTTPException(status_code=429, detail="Rate limit exceeded")
return {"message": "Success"}

4. Django Implementation with Redis

4.1 Installation

pip install django-ratelimit redis

4.2 Using django-ratelimit Decorator

from django.http import JsonResponse
from django_ratelimit.decorators import ratelimit
@ratelimit(key='ip', rate='100/h', method='GET')
def api_view(request):
return JsonResponse({"data": "value"})
@ratelimit(key='user', rate='5/m', method='POST')
def login_view(request):
# 5 login attempts per minute per user
return JsonResponse({"status": "logged in"})

4.3 Custom Redis Middleware for Django

import redis
import time
from django.http import JsonResponse
from django.utils.decorators import decorator_from_middleware
from django.utils.decorators import method_decorator
redis_client = redis.Redis(host='localhost', port=6379, db=0)
class RateLimitMiddleware:
def __init__(self, get_response):
self.get_response = get_response
self.rate_limit = 100
self.window = 60
def __call__(self, request):
identifier = request.META.get('REMOTE_ADDR')
key = f"rate_limit:{identifier}"
current = redis_client.get(key)
if current is None:
redis_client.setex(key, self.window, 1)
elif int(current) < self.rate_limit:
redis_client.incr(key)
else:
return JsonResponse(
{"error": "Rate limit exceeded"},
status=429
)
response = self.get_response(request)
response['X-RateLimit-Limit'] = self.rate_limit
response['X-RateLimit-Remaining'] = self.rate_limit - int(redis_client.get(key) or 0)
return response
# settings.py
MIDDLEWARE = [
# ... other middleware
'myapp.middleware.RateLimitMiddleware',
]

5. Redis Setup

5.1 Local Installation

# macOS
brew install redis
redis-server
# Linux
sudo apt-get install redis-server
redis-server
# Docker
docker run -d -p 6379:6379 redis:latest

5.2 Test Redis Connection

import redis
redis_client = redis.Redis(host='localhost', port=6379, db=0)
redis_client.ping()  # Returns True if connected

Reference: Redis Installation Guide

6. Advanced: Per-Endpoint Rate Limits

from fastapi import FastAPI, Depends, HTTPException
from functools import wraps
app = FastAPI()
def rate_limit(requests: int, window: int):
def decorator(func):
@wraps(func)
async def wrapper(request: Request, *args, **kwargs):
identifier = f"{request.client.host}:{func.__name__}"
key = f"rate_limit:{identifier}"
current = redis_client.get(key)
if current is None:
redis_client.setex(key, window, 1)
elif int(current) < requests:
redis_client.incr(key)
else:
raise HTTPException(status_code=429, detail="Rate limit exceeded")
return await func(request, *args, **kwargs)
return wrapper
return decorator
@app.get("/api/search")
@rate_limit(requests=30, window=60)
async def search(request: Request, q: str):
return {"results": []}
@app.post("/api/login")
@rate_limit(requests=5, window=900)  # 5 per 15 minutes
async def login(request: Request, credentials: dict):
return {"token": "xyz"}

7. Respond with Rate Limit Headers

Include rate limit info in response headers for client awareness:

from fastapi import FastAPI
from fastapi.responses import JSONResponse
@app.get("/api/data")
async def get_data(request: Request):
identifier = request.client.host
key = f"rate_limit:{identifier}"
current = redis_client.get(key) or 0
remaining = max(0, 100 - int(current))
return JSONResponse(
content={"data": "value"},
headers={
"X-RateLimit-Limit": "100",
"X-RateLimit-Remaining": str(remaining),
"X-RateLimit-Reset": str(int(time.time()) + 60)
}
)

Clients can check these headers to know how many requests they have left.

8. Testing Rate Limits

import pytest
from fastapi.testclient import TestClient
client = TestClient(app)
def test_rate_limit():
# Make 11 requests (limit is 10 per minute)
for i in range(10):
response = client.get("/api/data")
assert response.status_code == 200
# 11th request should fail
response = client.get("/api/data")
assert response.status_code == 429
assert "Rate limit" in response.json()["detail"]

9. Monitoring and Alerts

Track rate limit violations to detect attacks:

import logging
logger = logging.getLogger(__name__)
async def check_rate_limit(identifier: str):
key = f"rate_limit:{identifier}"
current = redis_client.get(key)
if current and int(current) > RATE_LIMIT:
logger.warning(f"Rate limit exceeded for {identifier}")
# Send alert to monitoring service
return False
redis_client.incr(key)
return True

Reference: slowapi Documentation

10. Best Practices Checklist

Use per-user rate limits, not just per-IP (authenticated endpoints)
Different limits for different endpoints (auth endpoints stricter)
Include rate limit info in response headers
Return clear 429 responses with retry-after headers
Monitor and alert on rate limit violations
Use sliding window or token bucket for accuracy
Test rate limiting before deployment
Consider distributed rate limiting for multi-server setups

11. Distributed Rate Limiting

For multiple servers, Redis is the source of truth:

# Server 1
redis_client.incr(f"rate_limit:user_123")
# Server 2 (checks same Redis instance)
current = redis_client.get(f"rate_limit:user_123")
# All servers see the same count

This ensures rate limits are enforced across your entire cluster, not per-server.