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!
Top comments (0)