Rate limiting is the simplest defense against abuse. Here's how to add it
to a FastAPI app in 2026, step by step.
Install
pip install drogue[fastapi]
Basic rate limiting
from fastapi import FastAPI
from drogue.adapters.fastapi import DrogueLimiter
app = FastAPI()
limiter = DrogueLimiter(app, default_limits=["100/minute"])
@app.get("/api/data")
@limiter.limit("10/minute")
async def get_data():
return {"data": "value"}
That's it. Every request gets a limit. Exceed it and you get 429 with
standard headers: X-RateLimit-Limit, X-RateLimit-Remaining, Retry-After.
Per-route limits
@app.get("/api/public")
@limiter.limit("100/minute")
async def public():
return {"ok": True}
@app.post("/api/login")
@limiter.limit("5/minute") # strict: brute force protection
async def login(credentials: dict):
return {"token": "..."}
@app.post("/api/upload")
@limiter.limit("10/minute")
async def upload(file: UploadFile):
return {"filename": file.filename}
No request: Request needed. drogue extracts identity from the ASGI
context automatically.
Choosing an algorithm
from drogue.core.rules.rule import AlgorithmType
@app.get("/api/data")
@limiter.limit("30/minute", algorithm=AlgorithmType.FIXED_WINDOW)
async def get_data():
return {"data": "value"}
| Algorithm | Behavior | Use when |
|---|---|---|
TOKEN_BUCKET (default) |
Allows short bursts above average rate | General API protection |
FIXED_WINDOW |
Hard cap per window, no bursts | Login, payment, strict quotas |
SLIDING_WINDOW |
Smooth rate limiting, no boundary spikes | Fair API usage |
GCRA |
Telecom-grade smooth traffic | High-throughput APIs |
LEAKY_BUCKET |
Constant-rate processing | Background jobs, queues |
Token Bucket is the default because it's burst-friendly. For brute force
protection on login endpoints, use FIXED_WINDOW.
Custom identity (not just IP)
from drogue.core.identity import HeaderExtractor
api_key_extractor = HeaderExtractor("X-API-Key")
@app.get("/api/data")
@limiter.limit("100/minute", key_func=api_key_extractor)
async def get_data():
return {"data": "value"}
Available extractors:
-
RemoteAddressExtractor(default) - client IP with X-Forwarded-For anti-spoof -
UserExtractor- user ID from request context -
HeaderExtractor- custom header value -
CompositeExtractor- combine multiple
Redis for multi-process
from drogue.core.storage.redis import RedisStorage
storage = RedisStorage(url="redis://localhost:6379")
limiter = DrogueLimiter(app, storage=storage, default_limits=["100/minute"])
With Redis, rate limits are shared across all workers and processes.
Custom 429 response
from starlette.responses import JSONResponse
def my_error(request, exc, headers):
return JSONResponse(
{"error": "slow_down", "retry_after": exc.retry_after},
status_code=429,
headers=headers,
)
limiter = DrogueLimiter(app, error_renderer=my_error)
Rate limit headers are injected automatically.
WebSocket rate limiting
@app.websocket("/ws/chat")
@limiter.limit_ws("10/minute")
async def chat(websocket: WebSocket):
await websocket.accept()
...
What's next
- DDoS detection - Z-score anomaly detection
- Auto-ban - progressive bans for repeat offenders
- Trust scoring - 9x throughput for verified users
- CAPTCHA challenges - Turnstile integration for page traffic
All off by default. Enable what you need.
pip install drogue[fastapi]
Top comments (0)