DEV Community

Govind Avireddi
Govind Avireddi

Posted on

8

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:

HTTP request flow with custom 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)


Enter fullscreen mode Exit fullscreen mode

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


Enter fullscreen mode Exit fullscreen mode

Happy coding!

Image of Timescale

🚀 pgai Vectorizer: SQLAlchemy and LiteLLM Make Vector Search Simple

We built pgai Vectorizer to simplify embedding management for AI applications—without needing a separate database or complex infrastructure. Since launch, developers have created over 3,000 vectorizers on Timescale Cloud, with many more self-hosted.

Read more

Top comments (0)

A Workflow Copilot. Tailored to You.

Pieces.app image

Our desktop app, with its intelligent copilot, streamlines coding by generating snippets, extracting code from screenshots, and accelerating problem-solving.

Read the docs