How to add Custom Middleware in FastAPI?

FastAPI is a popular Python-based framework primarily used to build APIs. While developing an API, there are often scenarios where we need to perform certain operations on every user’s request before it reaches the specific endpoints. To address this need, the concept of middleware was introduced.

Middleware is essentially a function—simple or complex—that executes for every HTTP request before it is forwarded to the given endpoint operation. It also runs similarly for the response after the route processing is complete.

In this article, we will explore how to write custom middleware in FastAPI. Custom middleware allows developers to execute their own logic before and after each HTTP request, following the DRY (Don’t Repeat Yourself) principle, and helps to enable extending the functionality of the API.

Why Write Custom Middleware in FastAPI?

• Reduce Code Repetition: Middleware helps avoid repeating the same logic across multiple endpoints.
• Consistent Behavior: It ensures the same behavior is applied to all endpoints.
• Request Filtering: Middleware can block unnecessary or invalid requests before they reach the API endpoints.
• Advanced Logic Implementation: It allows the implementation of advanced operations such as logging, authentication, or performance monitoring.

How Middleware Works

In simple terms, middleware is a function that executes in a specific order in FastAPI. When a user sends a request to an endpoint, the request is first intercepted by the middleware. If everything is valid, the request is forwarded to the API endpoint; otherwise, an error can be returned.

After the endpoint processes the request and generates a response, the response again passes through the middleware before being sent back to the client.

For example, if there are two middleware functions, A and B, and a user sends a request to the /home endpoint:

1. The request is first intercepted by A, then by B.
2. The endpoint executes and generates the response.
3. On its way back, the response is intercepted first by B, then A.

In short, middleware handles requests in the order they are added and processes responses in reverse order.

Creating Custom Middleware

FastAPI provides a simple mechanism to create middleware using the @app.middleware("http") decorator. This decorator wraps an asynchronous function that executes before and after each request.

Here’s a simple example:

 
 
from fastapi import FastAPI, Request
 
app = FastAPI()
 
@app.middleware("http")  
async def custom_middleware(request: Request, call_next):  
    # Runs before the request reaches the route  
    print("Before Request")
 
    response = await call_next(request)
 
    # Runs after the route processing  
    print("After Request")  
    return response
 
@app.get("/")  
async def root():  
    return {"message": "Hello FastAPI Custom Middleware Learner"}
 

In the above code:

• request is the incoming HTTP request containing information like headers, body, method, path, etc.
• call_next(request) forwards the request to the next middleware, if any, or to the API endpoint.
• response is the HTTP response returned by the endpoint or the next middleware. You can modify it as needed before sending it back to the client.

Creating Custom Middleware in a Separate File

You can organize your middleware in a separate file, such as middleware.py, and import it into your main application file. This approach improves code structure and maintainability.

middleware.py

 
from fastapi import Request  
from starlette.middleware.base import BaseHTTPMiddleware
 
class CustomMiddleware(BaseHTTPMiddleware):  
    def __init__(self, app):  
        super().__init__(app)
 
    async def dispatch(self, request: Request, call_next):  
        print("Before request")  
        response = await call_next(request)  
        print("After request")  
        return response
 

main.py

 
from fastapi import FastAPI  
from middleware import CustomMiddleware
 
app = FastAPI()
 
# Add the custom middleware  
app.add_middleware(CustomMiddleware)
 
@app.get("/")  
async def read_root():  
    return {"message": "Create Custom Middleware in Different File"}
 

In the above example:

• We extend BaseHTTPMiddleware from Starlette, as FastAPI itself does not provide a direct class-based middleware option. FastAPI is fully compatible with Starlette middleware.
• The CustomMiddleware class inherits from BaseHTTPMiddleware, and the parent class is initialized in __init__.
• The dispatch function contains the logic to be executed before and after the request.
• call_next(request) forwards the request to the next middleware or API endpoint.

With this setup, every HTTP request to the API passes through this custom middleware.

Implementing a Rate Limiting Middleware

To better understand custom middleware, let’s implement a Rate Limiting Middleware that limits the number of requests a client can make within a specific time window.

middleware.py

 
import time  
from fastapi import Request, status  
from fastapi.responses import JSONResponse  
from starlette.middleware.base import BaseHTTPMiddleware  
from collections import defaultdict
 
class RateLimitMiddleware(BaseHTTPMiddleware):  
    def __init__(self, app, max_requests: int = 5, time_window: int = 60):  
        super().__init__(app)  
        self.max_requests = max_requests  
        self.time_window = time_window  
        self.requests = defaultdict(list)
 
    async def dispatch(self, request: Request, call_next):  
        client_ip = request.client.host  
        current_time = time.time()
 
        # Filter out requests outside the time window  
        request_times = [t for t in self.requests[client_ip] if t > current_time - self.time_window]  
        self.requests[client_ip] = request_times
 
        # Check if rate limit exceeded  
        if len(request_times) >= self.max_requests:  
            return JSONResponse(  
                status_code=status.HTTP_429_TOO_MANY_REQUESTS,  
                content={"message": "Rate limit exceeded"}  
            )
 
        # Record the new request  
        request_times.append(current_time)  
        self.requests[client_ip] = request_times
 
        # Forward request to the next middleware or endpoint  
        response = await call_next(request)  
        return response
 

main.py

 
from fastapi import FastAPI  
from middleware import RateLimitMiddleware
 
app = FastAPI()
 
# Add rate limiting middleware  
app.add_middleware(RateLimitMiddleware, max_requests=5, time_window=30)
 
@app.get("/")  
async def root():  
    return {"message": "This endpoint is not rate-limited"}
 
@app.get("/data")  
async def get_data():  
    return {"data": "Here is your rate-limited data"}
 

Explanation:

• RateLimitMiddleware tracks requests per client IP using a dictionary.
• Requests outside the defined time_window are removed to maintain only recent requests.
• If a client exceeds max_requests within the time window, a 429 Too Many Requests response is returned.
• Otherwise, the request is forwarded to the endpoint or next middleware.

This approach ensures that endpoints are protected against abuse while demonstrating the power of custom middleware in FastAPI.

Conclusion

Middleware is a powerful approach for handling complex operations during and after HTTP requests and responses. Creating custom middleware not only extends its functionality but also keeps the code maintainable. It can enhance security, implement monitoring systems, and ensure consistent behavior across all API endpoints.