ASGI middleware: access response data from the self.app call

[kkroupa]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from fastapi import FastAPI
from starlette.types import ASGIApp, Receive, Send, Scope
from starlette.requests import Request


class BaseHTTPMiddleware:
    def __init__(self, app: ASGIApp):
        self.app = app

    @staticmethod
    def get_request_object(scope, receive: Receive, send: Send) -> Request:
        return Request(scope, receive, send)

    async def dispatch(self, request: Request):
        raise NotImplementedError()

    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        if scope.get("type") == "http":
            return await self.dispatch(self.get_request_object(scope, receive, send))
        else:
            return await self.app(scope, receive, send)


class DummyMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: ASGIApp):
        self.app = app

    @staticmethod
    def get_request_object(scope, receive: Receive, send: Send) -> Request:
        return Request(scope, receive, send)

    async def dispatch(self, request: Request):
        response = await self.app(request.scope, request.receive, request._send)

        # How do I access the status code, for example? Or the response body?
        assert response is not None  # --> False


app = FastAPI()

app.add_middleware(DummyMiddleware)


@app.get("/")
async def read_root():
    return {"Hello": "World"}

Description

I'm trying to access the request response data inside a custom http middleware that is supposed to shadow the original one from the Starlette library, but I'm unable to do it, the coroutine returns None - but I'm thinking there must be a way as the response gets to the client without any issues. Any tips, explanation or further links would be much appreciated.

Apologies in advance if this question should've been posted in Starlette's discussions, I wasn't sure.

Operating System

macOS

Operating System Details

No response

FastAPI Version

0.101.0

Pydantic Version

2.1.1

Python Version

3.11.3

Additional Context

No response

[Kludex]

It looks like you are missing concepts.

Maybe the Middleware page on the starlette documentation can help.

[kkroupa]

Thanks for the reply. Could you kindly tell me what concepts am I missing? I've searched for the answers in the documentation before considering posting questions here.

[Kludex]

Reading, and understand this page is far better for you: https://www.starlette.io/middleware/

[nwrahnema]

The docs are limited to saying

You can also reuse responses, which are ASGI applications as well.

with no further explanation. I understand that you can return a new Response as mentioned in "Sending eager responses" but would it be possible to construct a Response object from the messages passed to the send function?

From looking through BaseHTTPMiddleware it seems like the answer is no unless you want to copy a ton of code, so would be nice to add that caveat to the docs.

[arrmansa]

This is what I have been using, without constructing a new request object

        response = await call_next(request)
        content = [gen async for gen in response.body_iterator]
        async def list_to_async_iterator(body_list):
            for part in body_list:
                yield part
        response.body_iterator = list_to_async_iterator(content)
        logging.info(f"RESPONSE {response.status_code} {response.raw_headers} {(b''.join(content)).decode('latin-1')}")
        return response