How API Rate Limiting Works
API rate limiting is a method used to control the amount of API requests a user can make within a certain period of time. This is important for maintaining the performance and stability of the API server and preventing abuse or misuse of the API.
Implementation Examples in FastAPI
FastAPI is a modern, fast (high-performance), web framework for building APIs with Python 3.7+ based on standard Python type hints. Here are two examples of how API rate limiting can be implemented in FastAPI:
Example 1: Using built-in rate limiting middleware
FastAPI provides a built-in rate limiting middleware called RateLimiter
which can be used to limit the number of requests to specific routes or endpoints. This middleware allows you to set a rate limit, such as 100 requests per minute, and automatically handles throttling or rejecting requests that exceed the limit.
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.limiter import Limiter, RateLimiter
from fastapi_limiter import FastAPILimiter
app = FastAPI()
limiter = Limiter(key_func=get_remote_address, default_limits=["200 per minute"])
app.state.limiter = limiter
app.add_middleware(RateLimiter, limiter=limiter)
@app.get("/limited-route")
async def limited_route():
return {"message": "This route is limited to 200 requests per minute"}
FastAPILimiter(app, key_func=get_remote_address)
Example 2: Custom rate limiting with Redis
FastAPI allows you to implement custom rate limiting using external data storage such as Redis. In this example, we use the aioredis
library to interact with a Redis server and implement rate limiting by tracking and limiting the number of requests per user ID or IP address.
import aioredis
from fastapi import FastAPI
from fastapi_limiter import FastAPILimiter
from fastapi_limiter.depends import RateLimiter
app = FastAPI()
async def get_redis_conn():
redis = await aioredis.create_redis_pool("redis://localhost")
try:
yield redis
finally:
redis.close()
await redis.wait_closed()
@app.get("/custom-limited-route")
async def limited_route(redis: aioredis.Redis=Depends(get_redis_conn)):
rate_limit = RateLimiter(store=redis, key="client_ip", rate=100, expire=60)
await rate_limit.consume()
return {"message": "This route is limited to 100 requests per minute"}
FastAPILimiter(app)
By implementing API rate limiting in FastAPI, you can better control the usage of your API and ensure its performance and stability under heavy load. Whether using built-in middleware or custom implementations with external data storage like Redis, FastAPI provides the flexibility to enforce rate limits and protect your API.
This is using aioredis under the hood and give duplicate base class TimeoutError when using python version of 3.11+
thanks a bunch
Great content