Write a custom FastAPI middleware class

David Y.

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.

Loved by over 4 million developers and more than 90,000 organizations worldwide, Sentry provides code-level observability to many of the world’s best-known companies like Disney, Peloton, Cloudflare, Eventbrite, Slack, Supercell, and Rockstar Games. Each month we process billions of exceptions from the most popular products on the internet.

Share on Twitter
Bookmark this page
Ask a questionJoin the discussion

Related Answers

A better experience for your users. An easier life for your developers.

© 2024 • Sentry is a registered Trademark
of Functional Software, Inc.