slowapi
Rate limiting is a strategy for limiting network traffic. It restricts the number of requests that a client can make to a server within a specified time frame. There are different types of rate - limiting algorithms, such as fixed window, sliding window, and leaky bucket.
First, make sure you have FastAPI and Uvicorn installed. You can install them using pip
:
pip install fastapi uvicorn
Here is a simple FastAPI application:
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "World"}
To run this application, use the following command:
uvicorn main:app --reload
slowapi
slowapi
is a popular library for implementing rate limiting in Python web applications. Install it using pip
:
pip install slowapi
from fastapi import FastAPI
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 = FastAPI()
limiter.state.limiter._default_limits = ["5/minute"]
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.get("/")
@limiter.limit("5/minute")
def read_root():
return {"Hello": "World"}
In the above example, we applied a rate limit of 5 requests per minute to the root route (/
). The get_remote_address
function is used to identify clients based on their IP addresses.
As shown in the previous example, we can use the client’s IP address to enforce rate limits. This is useful for preventing abuse from a single IP address.
from fastapi import FastAPI, Request
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 = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.get("/ip - limited")
@limiter.limit("3/minute")
def ip_limited_route(request: Request):
return {"message": "This route is IP - based rate limited"}
If your application has user authentication, you can use the user ID to enforce rate limits.
from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
# Assume we have a function to get user ID
def get_user_id(request: Request):
# Replace this with actual user ID retrieval logic
return "user123"
limiter = Limiter(key_func=get_user_id)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.get("/user - limited")
@limiter.limit("2/minute")
def user_limited_route(request: Request):
return {"message": "This route is user - based rate limited"}
When a client exceeds the rate limit, it’s important to provide a meaningful response. The slowapi
library provides a default handler, but you can customize it.
from fastapi import FastAPI, Request
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 = FastAPI()
app.state.limiter = limiter
def custom_rate_limit_exceeded_handler(request: Request, exc: RateLimitExceeded):
return {"detail": f"Rate limit exceeded: {exc.detail}"}
app.add_exception_handler(RateLimitExceeded, custom_rate_limit_exceeded_handler)
@app.get("/custom - limit")
@limiter.limit("1/minute")
def custom_limit_route():
return {"message": "This route has a custom rate limit handler"}
For large - scale applications, you may need to use a distributed rate - limiting solution. Redis is a popular choice for implementing distributed rate limits. You can use libraries like slowapi - redis
to integrate Redis with slowapi
.
pip install slowapi - redis
from fastapi import FastAPI
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.middleware import SlowAPIMiddleware
from slowapi_redis import RedisStorage
import redis
redis_client = redis.Redis(host='localhost', port=6379, db=0)
storage = RedisStorage(redis_client)
limiter = Limiter(key_func=get_remote_address, storage=storage)
app = FastAPI()
app.add_middleware(SlowAPIMiddleware)
@app.get("/distributed - limit")
@limiter.limit("10/minute")
def distributed_limit_route():
return {"message": "This route uses distributed rate limiting"}
Rate limiting is an essential feature for any web application exposed to the public. In FastAPI, you can easily implement rate limiting using libraries like slowapi
. By understanding the fundamental concepts, common practices, and best practices, you can protect your application from abuse and ensure fair resource usage. Whether you are using IP - based or user - based rate limits, and whether you need to scale your rate - limiting solution, there are tools and techniques available to meet your needs.