Python FastAPI middleware to modify request and response

A "middleware" is a function that works with every request before it is processed by any specific path operation. And also with every response before returning it.

I am sharing this recipe because I struggled to find right information for myself on the internet for developing a custom FastAPI middleware that can modify the incoming request body (for POST, PUT) as well as the output response content.

There are several middlewares included in the FastAPI for common use cases. Check here for details. Starlette has lot more middlewares that work with FastAPI applications too. However if none of them meet your requirements then you can easily develop a custom middleware.

In this article I am going to share a boilerplate code for the custom middleware.

The example code shared here is written for JSON content type. But the code can be extended to support the other content mime-types.

The below diagram show the flow of HTTP requests from the client application which is intercepted by the middleware:

Here is the code for implementing this custom middleware:

import json
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
from starlette.types import Message

class MyMiddleware(BaseHTTPMiddleware):
    async def set_body(self, request: Request):
        receive_ = await request._receive()

        body = receive_.get('body')
        body['new_field'] = "new data" # modify the body here

        async def receive() -> Message:
            receive_["body"] = body
            return receive_

        request._receive = receive

    async def get_body(self, request: Request) -> bytes:
        body = await request.body()
        self.set_body(request)
        return body

    async def dispatch(self, request, call_next):

        if request.method == 'POST':
            self.set_body(request)

        response = await call_next(request)
        response_body = b""
        async for chunk in response.body_iterator:
            response_body += chunk
        response_json = json.loads(response_body.decode("utf-8"))
        response_json["new_field"] = "new data" # modify response here

        modified_response = json.dumps(response_json).encode("utf-8")
        response.headers['Content-Length'] = str(len(modified_response))
        return Response(content=json.dumps(response_json).encode("utf-8"), status_code=response.status_code,
            headers=dict(response.headers), media_type=response.media_type)

Now add this middleware to the FastAPI application as shown here:

from fastapi import FastAPI
from my_middleware import MyMiddleware

def create_app():
    app = FastAPI()

    # Add the middleware
    app.add_middleware(MyMiddleware)

    return app

Happy coding!