Skip to content

Accessing Responses from Dependencies

Xpresso gives you the ability to access and even modify responses from within dependencies. You will be able to:

  • Get a reference to the response returned by the endpoint function
  • Modify that response in place
  • Replace that response with a completely different response object

This functionality is enabled through response proxies:

  • xpresso.responses.get_response(request: Request) -> Response
  • xpresso.responses.set_response(request: Request, response: Response) -> None

These functions can only be called from within the teardown of a dependency. If called from anywhere else (inside the endpoint or in the setup of a context manager dependency) they will raise an exception. Further, modifying the response or calling set_response() will only work from a dependency in the "endpoint" scope (otherwise the response has already been sent).

Reading responses

Here is an example of a dependency that logs the status code for every response on a path:

from typing import Generator, List

from xpresso import (
    App,
    Depends,
    FromPath,
    HTTPException,
    Path,
    Request,
)
from xpresso.responses import get_response


class StatusCodeLogFile(List[int]):
    pass


def log_response_status_code(
    request: Request, log: StatusCodeLogFile
) -> Generator[None, None, None]:
    try:
        yield
    except HTTPException as exc:
        log.append(exc.status_code)
        raise
    else:
        response = get_response(request)
        log.append(response.status_code)


fake_items_db = {"foo": "Foo", "bar": "Bar"}


async def read_items(item_name: FromPath[str]) -> str:
    if item_name in fake_items_db:
        return fake_items_db[item_name]
    raise HTTPException(status_code=404)


app = App(
    routes=[
        Path(
            path="/items/{item_name}",
            get=read_items,
            dependencies=[
                Depends(log_response_status_code, scope="connection")
            ],
        ),
    ]
)

If your dependency has the "connection" scope (like in the example above) you will be able to get a copy of the request, but attempting to modify it or replace it will have no result since it was already sent to the client. The main advantage of using the "connection" scope is reduced latency for the client.

Writing responses

If you need to modify the response, use the "endpoint" scope. Here's an example of a simple request/context tracing system:

from typing import Generator
from uuid import UUID, uuid4

from xpresso import (
    App,
    Depends,
    FromPath,
    HTTPException,
    Path,
    Request,
)
from xpresso.responses import get_response

CONTEXT_HEADER = "X-Request-Context"


def trace(request: Request) -> Generator[None, None, None]:
    req_ctx = request.headers.get(CONTEXT_HEADER, None)
    if req_ctx is not None:
        ctx = UUID(req_ctx)
    else:
        ctx = uuid4()
    try:
        yield
    except HTTPException as exc:
        exc.headers[CONTEXT_HEADER] = str(ctx)
        raise
    else:
        response = get_response(request)
        response.headers[CONTEXT_HEADER] = str(ctx)


fake_items_db = {"foo": "Foo", "bar": "Bar"}


async def read_items(item_name: FromPath[str]) -> str:
    if item_name in fake_items_db:
        return fake_items_db[item_name]
    raise HTTPException(status_code=404)


app = App(
    routes=[
        Path(
            path="/items/{item_name}",
            get=read_items,
            dependencies=[Depends(trace, scope="endpoint")],
        ),
    ]
)