FastAPI middleware

Question

Is there any way to get raw response in middleware?

E.g.

app = FastAPI() 


@app.get("/any/route")
async def make_dummy_response():
    return [1,2,3]

In my case, middleware is good for processing all responses, but it need to decompress body response, that's produce an overhead.
How can I get raw return for all routes? In this case is list object.

Definitely, I can use a decorator, but they have to be added for each route which is inconvenient.

Middleware look like this

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    response = await call_next(request)
    # overhead
    response_body = [chunk async for chunk in response.body_iterator]
    response.body_iterator = iterate_in_threadpool(iter(response_body))

Have you seen https://stackoverflow.com/questions/69670125/how-to-log-raw-http-request-response-in-python-fastapi ? — MatsLindh, Sep 09 '22 at 08:38
Also, please have a look at [this answer](https://stackoverflow.com/questions/71882419/fastapi-how-to-get-the-response-body-in-middleware/71883126#71883126). — Chris, Sep 09 '22 at 10:40

score 1 · Answer 1 · answered Sep 09 '22 at 09:01

You can get entire response body in an ASGI middleware.

Sample code:

import typing

Scope = typing.MutableMapping[str, typing.Any]
Message = typing.MutableMapping[str, typing.Any]

Receive = typing.Callable[[], typing.Awaitable[Message]]
Send = typing.Callable[[Message], typing.Awaitable[None]]

ASGIApp = typing.Callable[[Scope, Receive, Send], typing.Awaitable[None]]


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

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            """
            this middleware only handles http scope.
            Ignore other ASGI compatible scopes (lifespan and websockets) here.
            """
            return await self.app(scope, receive, send)

        async def dummy_send(message: Message):
            if message.get("type", "") == "http.response.body":
                print(message.get("body"))

            _send = await send(message)
            return _send

        await self.app(scope, receive, dummy_send)

And you can use this in any ASGI framework.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

app.add_middleware(ASGIMiddleware)


class Model(BaseModel):
    text: str
    foo: str = None


@app.post("/")
def check(req: Model):
    req.foo = "set"
    return req

I also have to do json.loads(message.get("body").decode()) in this case. message.get("body") return bytes — RazDva, Sep 09 '22 at 11:52
Yes, this is barebones implementation. You need to decode if you want to work with dictionary. — Irfanuddin, Sep 09 '22 at 12:18

FastAPI middleware

1 Answers1