We can implement a rate limit for our calls using slowapi, which is the equivalent of Flask-Limiter but for FastAPI:
Install slowapi
pip install slowapi
Configure the limiter
# main.py from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
Apply the limit to the endpoints
# app/routers/users.py from slowapi import Limiter from slowapi.util import get_remote_address limiter = Limiter(key_func=get_remote_address) @router.get("/register") @limiter.limit("5/5seconds") async def register(request: Request, ...): ... @router.get("/our_endpoint") @limiter.limit("5/5seconds") async def our_endpoint(request: Request, ...): ...\
Use Redis as storage
By default it uses memory, but since you already have Redis:
limiter = Limiter( key_func=get_remote_address, storage_uri="redis://localhost:6379" )
Change localhost for your Redis container name if you are using Docker Compose.
With this configuration you will not be able to make more than 5 calls every 5 seconds.
Decorators available
# By second @limiter.limit("10/second") @limiter.limit("10/seconds") # By minute @limiter.limit("10/minute") @limiter.limit("10/minutes") # By hour @limiter.limit("10/hour") @limiter.limit("10/hours") # By day @limiter.limit("10/day") @limiter.limit("10/days") # Multiple at once (AND, applies all) @limiter.limit("5/second;100/minute;1000/day") # With variable @limiter.limit("10/minute", key_func=lambda r: r.headers.get("X-API-Key", get_remote_address(r)))
