Access ASGI scope in a HTTP middleware

[scy]

Hi there! I'd like to write a middleware that knows which endpoint function the request has been handled by.

This could be useful for example to collect some statistics, grouped by the __name__ of the endpoint function, or for logging/debugging purposes, or for selecting some kind of transformation (e.g. a template file) in a generic way based on the function, whatever.

I have learned that I can do this if I write a pure ASGI middleware, because after await self.app(scope, send, receive), scope["endpoint"] will contain the reference to the function.

However, if I then also want to do something with, for example, the Response object, I can't easily do it, because my middleware runs too low-level for that. I would need to hook into the send calls and so on, which is quite cumbersome.

Basically, what I want is the best of both worlds.

Now, Starlette's BaseHTTPMiddleware is a pure ASGI middleware, and the simplified interface it provides (subclass it, implement dispatch, be done) is really nice – but dispatch only receives references to request and call_next. There's no way for me to get access to scope from inside a dispatch implementation.

Would it be possible and make sense to

• either extend the dispatch interface to something like dispatch(self, request: Request, call_next: RequestResponseEndpoint, *, scope: Scope)
• or make a reference to the ASGI scope available on the Request and/or Response classes

so that there's a way for me access the ASGI scope from a BaseHTTPMiddleware?

But I've also seen that you're not exactly keen on encouraging people to use BaseHTTPMiddleware at all (#1729, #2160). So maybe your preferred option is for me to indeed write a pure ASGI middleware and deal with the additional complexity?

---

To give you an example: I've written FastAPI code that basically does content auto-negotiation; an endpoint usually returns JSON, but if Accept: text/html is preferred (or it's a HTMX request that expects a HTML fragment), the response is passed through Jinja instead to return HTML.

I've subclassed FastAPI's default JSONResponse to keep track of the original Python object, so that I don't have to parse the body's JSON again to give it to Jinja.

I've also implemented a clever way to make this completely transparent for the actual endpoint functions by trying to use a Jinja template named function_name.html, where function_name is the __name__ of the function that handled the request. But in order to do that, I had to implement a custom route handler, because a simple middleware doesn't have access to that information.

What do you think would be the best way to do something like this? Could I still access something like the original_data in PreserveJSONResponse with a pure ASGI middleware? Or would I have to re-parse the body? (Or … hmmm. Maybe I could even intercept the original Python object before it's being converted into JSON?)

from typing import Any, Callable

from fastapi import FastAPI, Request, Response
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.routing import APIRoute
from jinja2 import Environment, FileSystemLoader, Template, TemplateNotFound
from pydantic import BaseModel


# By default, FastAPI's JSONResponse will render the response and then forget
# the original data that has been passed in. We subclass the default behavior to
# keep the original Python object, so that we can later pass it into Jinja.


class PreserveJSONResponse(JSONResponse):
    """A JSONResponse that remembers the original Python object."""

    def render(self, content: Any) -> bytes:
        # Store the Python object in self.original_data.
        self.original_data = content
        # Use the parent class to convert to JSON.
        return super().render(content)


app = FastAPI(default_response_class=PreserveJSONResponse)


# We define a custom route handler to hook into FastAPI's request/response
# handling. This is because we want to be able to automatically select a
# template based on the path operation function that handled the request.
# Apparently a simple middleware sadly doesn't have access to that information.


def get_template_render_route_class(template_dir: str) -> type:
    """Return APIRoute subclass that renders HTML templates from JSON.

    This is a function that returns a class, mainly because we'd like to be able
    to customize the template_dir, but we don't call __init__() on the class
    ourself, FastAPI does it.
    """

    class RenderRoute(APIRoute):

        # This method will be called on startup to get the actual handler.
        def get_route_handler(self) -> Callable:
            # Get the handler from the parent class.
            original = super().get_route_handler()

            # Set up Jinja to be able to render templates.
            jinja_env = Environment(
                loader=FileSystemLoader(template_dir),
            )

            def is_hx(request: Request) -> bool:
                """Check whether the request is sent by HTMX."""
                # HTMX always sets this header, see
                # <https://htmx.org/docs/#request-headers>.
                return request.headers.get("HX-Request", "") == "true"

            def wants_html(request: Request) -> bool:
                """Check whether the client requested HTML."""
                # This is just a proof of concept, more solid `Accept` header
                # parsing would be required.
                return request.headers.get("Accept", "") == "text/html"

            def should_render(request: Request) -> bool:
                """Check whether this request should be rendered as HTML."""
                return is_hx(request) or wants_html(request)

            def get_template(endpoint: Callable|None) -> Template|None:
                """Get Jinja template instance for a given path operation."""
                if endpoint is None:
                    # We don't know the function that handled this request,
                    # therefore we can't select a corresponding template.
                    return None

                # Simply add `.html` to the function name and try to look this
                # up as a template file.
                try:
                    return jinja_env.get_template(f"{endpoint.__name__}.html")
                except TemplateNotFound:
                    return None

            # This is the actual function that is called on every request.
            async def route_handler(request: Request) -> Response:
                # Handle the request and get back a response.
                response = await original(request)

                if (
                    # rendering as HTML makes sense
                    should_render(request)
                    # the response is actually JSON that we can work with
                    and isinstance(response, PreserveJSONResponse)
                    # we have a template available
                    and (template := get_template(self.dependant.call))
                ):
                    # Render the Jinja template and return as HTML response.
                    return HTMLResponse(template.render(
                        data=response.original_data,
                    ))

                # If anything failed or the requirements were not met, return
                # the original response.
                return response

            # Provide FastAPI with the route handler.
            return route_handler

    return RenderRoute


app.router.route_class = get_template_render_route_class("templates")


# The following is the actual FastAPI "application". Note that it doesn't know
# anything about the HTML rendering, it's being done completely transparent to
# the application's code.


class Item(BaseModel):
    """Trivial model for something that has a name."""
    name: str


@app.get("/items")
def get_items():
    """Return a list of items."""
    return [
        Item(name="Alpha"),
        Item(name="Beta"),
        Item(name="Gamma"),
    ]

[adriangb]

Can't you do request.scope?

[scy]

Duh. This totally works, thank you.

I think my mistake was that I inspected request.scope before the await call_next(request), at which point it doesn't have an endpoint attribute yet. Which makes total sense, because routing did not happen yet. 🤦‍♂️

Sorry for the noise, and thanks again!