Write a custom FastAPI middleware class

The Problem

I would like to create a custom middleware class for FastAPI that logs the duration of each request, for performance profiling purposes. This middleware will ultimately form part of a shared library between different FastAPI projects, so I would also like to avoid tying it to the code of any individual FastAPI project, as appears to be the approach in the FastAPI middleware documentation.

How do I accomplish this? Is middleware the right approach?

The Solution

In FastAPI, middleware processes requests before they’re sent to specific path operations and processes responses before they’re returned. This makes it ideal for generic operations we want to do on every request and response, such as logging the time between a request and its response.

To create a custom middleware, we must define a class that adheres to the ASGI (Asynchronous Server Gateway Interface) specification. Here’s an example of a middleware class for our timing operation, which we might store in a self-contained file named timing_middleware.py:

import time

class TimingMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        start_time = time.time()
        await self.app(scope, receive, send)
        duration = time.time() - start_time
        print(f"Request duration: {duration:.2f} seconds")

The important part of this class is the __call__ method, which implements the ASGI application specification. It is an async function that takes the parameters scope, receive and send. These parameters are needed for the FastAPI request. In this simple middleware, we just pass them through to our FastAPI app in the line await self.app(scope, receive, send). We can think of __call__ as a wrapper around our app’s requests.

Since this is a generic ASGI middleware class, we can use it not just with different FastAPI projects, but also with projects using other Python ASGI servers, such as Starlette, on which FastAPI is built.

Here’s how we might use this class in a FastAPI project:

from fastapi import FastAPI
from timing_middleware import TimingMiddleware # import middleware class

app = FastAPI()
app.add_middleware(TimingMiddleware) # add middleware to FastAPI app

@app.get("/hello")
async def greeter():
    return {"Hello": "World"}

@app.get("/goodbye")
async def farewell():
    return {"Goodbye": "World"}

Now our middleware will run every time we request one of the API’s defined routes. We should keep this in mind when extending the middleware’s capabilities – if our middleware’s __call__ function contains too much processing, it will slow down our entire application.