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] 
