FastAPI has quickly become a go-to framework for building high-performance APIs with Python. One of the features that makes FastAPI so powerful is its middleware system. Middleware allows you to run code before or after each request, making it incredibly useful for logging, authentication, and request/response transformation tasks.
In this article, we'll explore what middleware is in FastAPI, why it's essential, and how you can create your custom middleware. We'll even build a practical example of rate-limiting middleware to give you a hands-on understanding of how it works.
🚀 What is Middleware in FastAPI?
Middleware is essentially a layer that wraps around the request and response cycle. It's a function that runs before and after every request processed by your application. Think of it as a way to inject custom logic into the lifecycle of each request.
Here are some common use cases for middleware:
- Authentication & Authorization: Ensure that only authenticated users can access certain endpoints.
- Logging: Track incoming requests and outgoing responses for debugging or monitoring.
- Rate Limiting: Control the number of requests a client can make within a specific period.
- Request/Response Transformation: Modify requests before they reach your route handlers or responses before they're sent back to the client.
🛠️ How to Create Custom Middleware in FastAPI
Creating middleware in FastAPI is straightforward. You define a class that implements the middleware logic, and then you add it to your FastAPI app.
Step 1: Define the Middleware Class
Start by creating a Python class that accepts the FastAPI app instance. Inside this class, you'll define a dispatch method where you can add your custom logic.
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
class CustomMiddleware(BaseHTTPMiddleware):
def __init__(self, app):
super().__init__(app)
async def dispatch(self, request: Request, call_next):
# Perform actions before the request is processed
print("Before request")
# Forward the request to the next middleware or route handler
response = await call_next(request)
# Perform actions after the request is processed
print("After request")
return response
Step 2: Add the Middleware to Your FastAPI App
Once you've defined your middleware, the next step is to integrate it with your FastAPI application.
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
class CustomMiddleware(BaseHTTPMiddleware):
def __init__(self, app):
super().__init__(app)
async def dispatch(self, request: Request, call_next):
# Perform actions before the request is processed
print("Before request")
# Forward the request to the next middleware or route handler
response = await call_next(request)
# Perform actions after the request is processed
print("After request")
return response
With this setup, every request to your FastAPI app will pass through your custom middleware. This allows you to execute code before and after each request is handled by your route handlers.
🧑💻 Example: Implementing Rate Limiting Middleware
Rate limiting is a common requirement for APIs to prevent abuse and ensure fair usage. Let's walk through how to create a rate-limiting middleware in FastAPI.
from fastapi import Depends
from fastapi import Request, HTTPException
from time import time
from collections import defaultdict
class RateLimiter:
def __init__(self, max_requests: int, time_window: int):
self.max_requests = max_requests
self.time_window = time_window
self.requests = defaultdict(list)
def __call__(self, request: Request):
client_ip = request.client.host
current_time = time()
# Get the list of request times for the client IP
request_times = self.requests[client_ip]
# Remove outdated requests outside the time window
request_times = [timestamp for timestamp in request_times if timestamp > current_time - self.time_window]
self.requests[client_ip] = request_times
# Check if the client exceeds the max request limit
if len(request_times) >= self.max_requests:
raise HTTPException(status_code=429, detail="Rate limit exceeded")
# Add the current request time to the list
request_times.append(current_time)
# Initialize FastAPI app without global rate limiting
app = FastAPI()
# Create a RateLimiter instance
limiter = RateLimiter(max_requests=5, time_window=30)
@app.get("/")
async def root():
return {"message": "This endpoint is not rate-limited"}
@app.get("/data", dependencies=[Depends(limiter)])
async def get_data():
return {"data": "Here is your rate-limited data"}
💡 Conclusion
Middleware in FastAPI provides a powerful mechanism to handle cross-cutting concerns like logging, authentication, and rate limiting. By creating custom middleware, you can extend the capabilities of your FastAPI applications in a reusable and maintainable way.
Whether you're looking to enhance security, monitor application performance, or control access to your APIs, middleware gives you the tools to get the job done.
To explore more examples and see the code in action, you can check out my GitHub repository dedicated to FastAPI middleware.
I hope this guide helps you get started with creating and using middleware in FastAPI. Have any questions or ideas for middleware you'd like to build? Drop a comment below!
Top comments (0)