CODEWITHSUNDEEP
pythonmiddlewarefastapipython-decorators
RazDva
RazDva

Reputation: 114

FastAPI middleware

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))

Upvotes: 1

Views: 2826

Answers (1)

Irfanuddin
Irfanuddin

Reputation: 2605

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

Upvotes: 1

Related Questions