Implementing API Rate Limiting in Python: A Guide with FastAPI and Redis#

Rate Limiting Algorithms#

Several algorithms exist for defining and enforcing rate limits:

• Fixed Window: This is the simplest approach. A time window (e.g., 60 seconds) and a maximum request count are defined. All requests within that window are counted. Once the limit is reached, no more requests are allowed until the next window begins. A drawback is potential traffic bursts at the start of each window.
• Sliding Window Log: This method keeps a timestamped log of requests within a window. For each incoming request, it counts the number of requests in the log that fall within the current time window. While accurate, it can be memory-intensive for large volumes of requests.
• Sliding Window Counter: This popular method uses a fixed window but averages the rate over the previous window to smooth out bursts. It’s less accurate than the sliding window log but more memory-efficient.
• Token Bucket: This algorithm visualizes rate limiting as a bucket holding “tokens.” Requests consume tokens. Tokens are added to the bucket at a fixed rate. If the bucket is empty, requests are denied. This handles bursts better than fixed windows as long as there are tokens available.
• Leaky Bucket: This algorithm models a bucket with a leak at a constant rate. Requests fill the bucket. If the bucket is full, requests are denied. This smooths out bursts into a steady flow but can lead to requests being delayed if the bucket is not full but the incoming rate exceeds the leak rate.

Redis is particularly well-suited for implementing Fixed Window and Sliding Window Counter algorithms due to its atomic increment operations and key expiration capabilities.

Identifying the Client#

To enforce rate limits, the API needs to identify the entity making the request. Common identifiers include:

• IP Address: Simple to implement, but multiple users behind a single NAT share an IP, and a single user might have multiple IPs (IPv4/IPv6).
• API Key/Authentication Token: More accurate for identifying specific users or applications, especially for authenticated endpoints. This is generally preferred for rate limiting specific users or subscription tiers.
• User ID: Similar to API keys, suitable for authenticated users.

Step-by-Step Implementation with fastapi-limiter#

This guide uses fastapi-limiter which relies on Redis as the backend.

1. Installation: Install FastAPI, Uvicorn (an ASGI server), redis-py (Redis client), and fastapi-limiter:

   1
   pip install fastapi uvicorn redis fastapi-limiter
   

2. Connect to Redis: Establish a connection to your Redis instance. This typically happens at the application startup.

   1
   import redis.asyncio as redis
   2
   from fastapi import FastAPI
   3
   from fastapi_limiter import FastAPILimiter
   4
   
   5
   app = FastAPI()
   6
   redis_client: redis.Redis = None
   7
   
   8
   @app.on_event("startup")
   9
   async def startup():
   10
       # Replace with your Redis connection details
   11
       global redis_client
   12
       redis_client = redis.from_url("redis://localhost:6379", encoding="utf-8", decode_responses=True)
   13
       await FastAPILimiter.init(redis_client)
   14
   
   15
   @app.on_event("shutdown")
   16
   async def shutdown():
   17
       if redis_client:
   18
           await redis_client.close()
   

   • redis.asyncio is used for asynchronous Redis operations, aligning with FastAPI’s asynchronous nature.
   • The connection is initialized during the startup event and closed during the shutdown event.
   • FastAPILimiter.init() initializes the rate limiter with the Redis client.

3. Apply Rate Limits to Endpoints: Use the @limiter.limit() decorator provided by fastapi-limiter on your route handlers.

   1
   from fastapi import FastAPI
   2
   from fastapi_limiter.depends import RateLimiter
   3
   
   4
   # ... (previous startup/shutdown and redis_client definition) ...
   5
   
   6
   @app.get("/items")
   7
   @limiter.limit("5/minute") # Apply a rate limit of 5 requests per minute
   8
   async def read_items():
   9
       return [{"item": "Foo"}, {"item": "Bar"}]
   10
   
   11
   @app.post("/create_item")
   12
   @limiter.limit("1/second") # Apply a stricter limit of 1 request per second
   13
   async def create_item(item: dict):
   14
       return {"message": "Item created", "data": item}
   

   • @limiter.limit("5/minute") applies a rate limit. The string format specifies the rate (e.g., “5 requests per minute”). Other examples: "10/hour", "100/day".
   • fastapi-limiter uses the client’s IP address by default to identify the client for rate limiting.

4. Customize Key Function (Optional): By default, fastapi-limiter uses the client’s IP address as the rate limit key. You can define a custom function to generate a key based on other criteria, such as a user ID from an authentication token.

   1
   from fastapi import FastAPI, Depends, Request
   2
   from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
   3
   from fastapi_limiter.depends import RateLimiter
   4
   from typing import Optional
   5
   
   6
   # Assume get_current_user dependency exists and returns a user object with an 'id' attribute
   7
   # from .dependencies import get_current_user
   8
   
   9
   # Dummy dependency for demonstration
   10
   async def get_current_user(credentials: Optional[HTTPAuthorizationCredentials] = Depends(HTTPBearer(auto_error=False))):
   11
       if credentials:
   12
           # In a real app, validate token and fetch user
   13
           # For demo, let's assume token "user123_token" maps to user_id "user123"
   14
           if credentials.credentials == "user123_token":
   15
                class DummyUser: # Simulate a user object
   16
                     def __init__(self, user_id):
   17
                          self.id = user_id
   18
                return DummyUser("user123")
   19
       return None # No user if no valid token
   20
   
   21
   def user_id_key_func(request: Request, user: Optional[object] = Depends(get_current_user)):
   22
       """Generates a rate limit key based on user ID or IP if not authenticated."""
   23
       if user and hasattr(user, 'id'):
   24
           return str(user.id)
   25
       return request.client.host # Fallback to IP if no user
   26
   
   27
   @app.get("/user_data")
   28
   # Apply limit based on the custom key function
   29
   @limiter.limit("10/minute", key_func=user_id_key_func)
   30
   async def read_user_data(user: Optional[object] = Depends(get_current_user)):
   31
        if user:
   32
             return {"message": f"Data for user {user.id}"}
   33
        return {"message": "Data for anonymous user"}
   

   • The key_func is an async or sync function that accepts the standard FastAPI dependency parameters (Request, Response, dependencies, etc.) and returns a string used as the Redis key suffix.
   • This example demonstrates creating a key based on a user ID if authenticated, or falling back to the IP address for unauthenticated requests.

5. Handling Rate Limit Exceptions: When a client exceeds the rate limit, fastapi-limiter raises a RateLimitException. You can add an exception handler to return a specific response.

   1
   from fastapi import FastAPI, Request
   2
   from fastapi.responses import JSONResponse
   3
   from fastapi_limiter.exceptions import RateLimitException
   4
   
   5
   # ... (previous startup/shutdown and redis_client definition) ...
   6
   
   7
   @app.exception_handler(RateLimitException)
   8
   async def rate_limit_exception_handler(request: Request, exc: RateLimitException):
   9
       return JSONResponse(
   10
           status_code=429, # Too Many Requests
   11
           content={"detail": "Rate limit exceeded"},
   12
       )
   

   • A custom exception handler catches RateLimitException and returns a 429 Too Many Requests HTTP status code, which is the standard response for rate-limited requests.

Running the Example#

Save the code as main.py and run with Uvicorn:

1
uvicorn main:app --reload

Requests to /items and /create_item will now be rate-limited based on the specified rules and the client’s IP address (or custom key function if implemented).