Creating Middleware

As mentioned in using middleware, a middleware in Litestar is any callable that takes a kwarg called app, which is the next ASGI handler, i.e. an ASGIApp, and returns an ASGIApp.

The example previously given was using a factory function, i.e.:

from litestar.types import ASGIApp, Scope, Receive, Send


def middleware_factory(app: ASGIApp) -> ASGIApp:
    async def my_middleware(scope: Scope, receive: Receive, send: Send) -> None:
        # do something here
        ...
        await app(scope, receive, send)

    return my_middleware

Extending ASGIMiddleware

While using functions is a perfectly viable approach, the recommended way to handle this is by using the ASGIMiddleware abstract base class, which also includes functionality to dynamically skip the middleware based on ASGI scope["type"], handler opt keys or path patterns and a simple way to pass configuration to middlewares; It does not implement an __init__ method, so subclasses are free to use it to customize the middleware’s configuration.

Modifying Requests and Responses

Middlewares can not only be used to execute around other ASGI callable, they can also intercept and modify both incoming and outgoing data in a request / response cycle by “wrapping” the respective receive and send ASGI callables.

The following demonstrates how to add a request timing header with a timestamp to all outgoing responses:

import time

from litestar.datastructures import MutableScopeHeaders
from litestar.enums import ScopeType
from litestar.middleware import ASGIMiddleware
from litestar.types import ASGIApp, Message, Receive, Scope, Send


class ProcessTimeHeader(ASGIMiddleware):
    scopes = (ScopeType.HTTP, ScopeType.ASGI)

    async def handle(self, scope: Scope, receive: Receive, send: Send, next_app: ASGIApp) -> None:
        start_time = time.monotonic()

        async def send_wrapper(message: Message) -> None:
            if message["type"] == "http.response.start":
                process_time = time.monotonic() - start_time
                headers = MutableScopeHeaders.from_message(message=message)
                headers["X-Process-Time"] = str(process_time)
            await send(message)

        await next_app(scope, receive, send_wrapper)

Configuration constraints

While it’s good practice to keep middlewares decoupled from another, there are times where implicit coupling is unavoidable due to the nature of the functionality provided by the middlewares.

For example a caching middleware and an authentication middleware can produce very different results depending on the order they are applied in; Assuming a naive caching middleware that does not take authentication state into account, if it’s applied before the authentication middleware, it might cache an authenticated response and serve it to the next, unauthenticated request.

Especially when applications grow larger and more complex, it can become difficult to keep track of all these implicit couplings and dependencies, or downright impossible if the middleware is implemented in a separate package and has no knowledge about how it is being applied.

To help with this, ASGIMiddleware allows to specify a set of MiddlewareConstraints - Once configured, these will be validated on application startup.

Using constraints, the example given above might be solved like this:

from litestar.middleware.authentication import AbstractAuthenticationMiddleware
from litestar.middleware.base import ASGIMiddleware
from litestar.middleware.constraints import MiddlewareConstraints


class CachingMiddleware(ASGIMiddleware):
    constraints = MiddlewareConstraints(after=(AbstractAuthenticationMiddleware,))

Here, we specify that every instance of CachingMiddleware must come after any instance of AbstractAuthenticationMiddleware.

Tip

When referencing classes, the constraints always apply to all instances and subclasses of the type

Forward references

Constraints that reference other middleware can use strings as forward references, to handle situations like circular imports or middlewares from packages that may not be available:

from litestar.middleware.base import ASGIMiddleware
from litestar.middleware.constraints import MiddlewareConstraints


class SomeMiddleware(ASGIMiddleware):
    constraints = MiddlewareConstraints().apply_after(
        "some_package.some_module.SomeMiddleware",
        ignore_import_error=True,
    )

This forward reference will try to import SomeMiddleware from some_package.some_module. With ignore_import_error=True, if the import is not successful, the constraint will be ignored.

Middleware order

For order constraints (before, after, first, last), it is important to note that the order is defined in terms of proximity to the location. In practice, this means that a middleware that has set first=True must be the first middleware on the first layer (i.e. the application), and a middleware setting last=True must be the last middleware on the last layer (i.e. the route handler).

@get("/", middleware=[FifthMiddleware, SixthMiddleware])
async def handler() -> None:
    pass

router = Router(
    "/",
    [handler],
    middleware=[
        ThirdMiddleware(),
        FourthMiddleware()
    ]
)

app = Litestar(
    middleware=[
        FirstMiddleware(),
        SecondMiddleware()
    ]
)

Constraints and plugins

When using plugins that add middleware(s), it is important to understand that these middleware(s) are added after middlewares defined on the application and before middlewares defined on the other layers.

Constraints are evaluated after all middlewares have been added though, so an order constraint on a middleware added by a plugin has to take into account the position it is being added to.

Most of the time in a plugin you would do app_config.middleware.append(MyCustomMiddleware) which will be ok if it has no constraints attached.

Now suppose that MyCustomMiddleware has the constraint first=True then the correct way to add it is to do app_config.middleware.insert(0, MyCustomMiddleware()) so that it is the first middleware in the stack.

Migrating from MiddlewareProtocol / AbstractMiddleware

ASGIMiddleware was introduced in Litestar 2.15. If you’ve been using MiddlewareProtocol / AbstractMiddleware to implement your middlewares before, there’s a simple migration path to using ASGIMiddleware.

From MiddlewareProtocol

from litestar.middleware import MiddlewareProtocol
from litestar.types import ASGIApp, Receive, Scope, Send


class MyMiddleware(MiddlewareProtocol):
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        # do stuff
        await self.app(scope, receive, send)

from litestar.middleware import ASGIMiddleware
from litestar.types import ASGIApp, Receive, Scope, Send


class MyMiddleware(ASGIMiddleware):
    async def handle(self, scope: Scope, receive: Receive, send: Send, next_app: ASGIApp) -> None:
        # do stuff
        await next_app(scope, receive, send)

From AbstractMiddleware

import anyio

from litestar import Litestar
from litestar.middleware import AbstractMiddleware, DefineMiddleware
from litestar.types import ASGIApp, Receive, Scope, Scopes, Send


class TimeoutMiddleware(AbstractMiddleware):
    def __init__(
        self,
        app: ASGIApp,
        timeout: float,
        exclude: str | list[str] | None = None,
        exclude_opt_key: str | None = None,
        scopes: Scopes | None = None,
    ):
        self.timeout = timeout
        super().__init__(app=app, exclude=exclude, exclude_opt_key=exclude_opt_key, scopes=scopes)

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        with anyio.move_on_after(self.timeout):
            await self.app(scope, receive, send)


app = Litestar(
    middleware=[
        DefineMiddleware(
            TimeoutMiddleware,
            timeout=5,
        )
    ]
)

import anyio

from litestar import Litestar
from litestar.middleware import ASGIMiddleware
from litestar.types import ASGIApp, Receive, Scope, Send


class TimeoutMiddleware(ASGIMiddleware):
    def __init__(self, timeout: float):
        self.timeout = timeout

    async def handle(self, scope: Scope, receive: Receive, send: Send, next_app: ASGIApp) -> None:
        with anyio.move_on_after(self.timeout):
            await next_app(scope, receive, send)


app = Litestar(middleware=[TimeoutMiddleware(timeout=5)])

Using MiddlewareProtocol

The MiddlewareProtocol class is a PEP 544 Protocol that specifies the minimal implementation of a middleware as follows:

from typing import Protocol, Any
from litestar.types import ASGIApp, Scope, Receive, Send


class MiddlewareProtocol(Protocol):
    def __init__(self, app: ASGIApp, **kwargs: Any) -> None: ...

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: ...

The __init__ method receives and sets “app”. It’s important to understand that app is not an instance of Litestar in this case, but rather the next middleware in the stack, which is also an ASGI app.

The __call__ method makes this class into a callable, i.e. once instantiated this class acts like a function, that has the signature of an ASGI app: The three parameters, scope, receive, send are specified by the ASGI specification, and their values originate with the ASGI server (e.g. uvicorn) used to run Litestar.

To use this protocol as a basis, simply subclass it - as you would any other class, and implement the two methods it specifies:

import logging

from litestar.types import ASGIApp, Receive, Scope, Send
from litestar import Request
from litestar.middleware.base import MiddlewareProtocol

logger = logging.getLogger(__name__)


class MyRequestLoggingMiddleware(MiddlewareProtocol):
    def __init__(self, app: ASGIApp) -> None:  # can have other parameters as well
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] == "http":
            request = Request(scope)
            logger.info("Got request: %s - %s", request.method, request.url)
        await self.app(scope, receive, send)

Important

Although scope is used to create an instance of request by passing it to the Request constructor, which makes it simpler to access because it does some parsing for you already, the actual source of truth remains scope - not the request. If you need to modify the data of the request you must modify the scope object, not any ephemeral request objects created as in the above.

Responding using the MiddlewareProtocol

Once a middleware finishes doing whatever its doing, it should pass scope, receive, and send to an ASGI app and await it. This is what’s happening in the above example with: await self.app(scope, receive, send). Let’s explore another example - redirecting the request to a different url from a middleware:

from litestar.types import ASGIApp, Receive, Scope, Send

from litestar.response.redirect import ASGIRedirectResponse
from litestar import Request
from litestar.middleware.base import MiddlewareProtocol


class RedirectMiddleware(MiddlewareProtocol):
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if Request(scope).session is None:
            response = ASGIRedirectResponse(path="/login")
            await response(scope, receive, send)
        else:
            await self.app(scope, receive, send)

As you can see in the above, given some condition (request.session being None) we create a ASGIRedirectResponse and then await it. Otherwise, we await self.app

Modifying ASGI Requests and Responses using the MiddlewareProtocol

Important

If you’d like to modify a Response object after it was created for a route handler function but before the actual response message is transmitted, the correct place to do this is using the special life-cycle hook called after_request. The instructions in this section are for how to modify the ASGI response message itself, which is a step further in the response process.

Using the MiddlewareProtocol you can intercept and modifying both the incoming and outgoing data in a request / response cycle by “wrapping” that respective receive and send ASGI functions.

To demonstrate this, let’s say we want to append a header with a timestamp to all outgoing responses. We could achieve this by doing the following:

import time

from litestar.datastructures import MutableScopeHeaders
from litestar.types import Message, Receive, Scope, Send
from litestar.middleware.base import MiddlewareProtocol
from litestar.types import ASGIApp


class ProcessTimeHeader(MiddlewareProtocol):
    def __init__(self, app: ASGIApp) -> None:
        super().__init__(app)
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] == "http":
            start_time = time.monotonic()

            async def send_wrapper(message: Message) -> None:
                if message["type"] == "http.response.start":
                    process_time = time.monotonic() - start_time
                    headers = MutableScopeHeaders.from_message(message=message)
                    headers["X-Process-Time"] = str(process_time)
                await send(message)

            await self.app(scope, receive, send_wrapper)
        else:
            await self.app(scope, receive, send)

Inheriting AbstractMiddleware

Litestar offers an AbstractMiddleware class that can be extended to create middleware:

import time

from litestar.enums import ScopeType
from litestar.middleware import AbstractMiddleware
from litestar.datastructures import MutableScopeHeaders
from litestar.types import Message, Receive, Scope, Send


class MyMiddleware(AbstractMiddleware):
    scopes = {ScopeType.HTTP}
    exclude = ["first_path", "second_path"]
    exclude_opt_key = "exclude_from_middleware"

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        start_time = time.monotonic()

        async def send_wrapper(message: "Message") -> None:
            if message["type"] == "http.response.start":
                process_time = time.monotonic() - start_time
                headers = MutableScopeHeaders.from_message(message=message)
                headers["X-Process-Time"] = str(process_time)
            await send(message)

        await self.app(scope, receive, send_wrapper)

The three class variables defined in the above example scopes, exclude, and exclude_opt_key can be used to fine-tune for which routes and request types the middleware is called:

  • The scopes variable is a set that can include either or both : ScopeType.HTTP and ScopeType.WEBSOCKET , with the default being both.

  • exclude accepts either a single string or list of strings that are compiled into a regex against which the request’s path is checked.

  • exclude_opt_key is the key to use for in a route handler’s Router.opt dict for a boolean, whether to omit from the middleware.

Thus, in the following example, the middleware will only run against the handler called not_excluded_handler for /greet route:

import time

from litestar import Litestar, WebSocket, get, websocket
from litestar.datastructures import MutableScopeHeaders
from litestar.enums import ScopeType
from litestar.middleware import ASGIMiddleware
from litestar.types import ASGIApp, Message, Receive, Scope, Send


class MyMiddleware(ASGIMiddleware):
    scopes = (ScopeType.HTTP,)
    exclude_path_pattern = ("first_path", "second_path")
    exclude_opt_key = "exclude_from_my_middleware"

    async def handle(self, scope: Scope, receive: Receive, send: Send, next_app: ASGIApp) -> None:
        start_time = time.monotonic()

        async def send_wrapper(message: "Message") -> None:
            if message["type"] == "http.response.start":
                process_time = time.monotonic() - start_time
                headers = MutableScopeHeaders.from_message(message=message)
                headers["X-Process-Time"] = str(process_time)
            await send(message)

        await next_app(scope, receive, send_wrapper)


@websocket("/my-websocket")
async def websocket_handler(socket: WebSocket) -> None:
    """
    Websocket handler - is excluded because the middleware scopes includes 'ScopeType.HTTP'
    """
    await socket.accept()
    await socket.send_json({"hello": "websocket"})
    await socket.close()


@get("/first_path", sync_to_thread=False)
def first_handler() -> dict[str, str]:
    """Handler is excluded due to regex pattern matching "first_path"."""
    return {"hello": "first"}


@get("/second_path", sync_to_thread=False)
def second_handler() -> dict[str, str]:
    """Handler is excluded due to regex pattern matching "second_path"."""
    return {"hello": "second"}


@get("/third_path", exclude_from_my_middleware=True, sync_to_thread=False)
def third_handler() -> dict[str, str]:
    """Handler is excluded due to the opt key 'exclude_from_my_middleware' matching the middleware 'exclude_opt_key'."""
    return {"hello": "third"}


@get("/greet", sync_to_thread=False)
def not_excluded_handler() -> dict[str, str]:
    """This handler is not excluded, and thus the middleware will execute on every incoming request to it."""
    return {"hello": "world"}


app = Litestar(
    route_handlers=[
        websocket_handler,
        first_handler,
        second_handler,
        third_handler,
        not_excluded_handler,
    ],
    middleware=[MyMiddleware()],
)

Danger

Using / as an exclude pattern, will disable this middleware for all routes, since, as a regex, it matches every path

Using DefineMiddleware to pass arguments

Litestar offers a simple way to pass positional arguments (*args) and keyword arguments (**kwargs) to middleware using the DefineMiddleware class. Let’s extend the factory function used in the examples above to take some args and kwargs and then use DefineMiddleware to pass these values to our middleware:

from litestar.types import ASGIApp, Scope, Receive, Send
from litestar import Litestar
from litestar.middleware import DefineMiddleware


def middleware_factory(my_arg: int, *, app: ASGIApp, my_kwarg: str) -> ASGIApp:
    async def my_middleware(scope: Scope, receive: Receive, send: Send) -> None:
        # here we can use my_arg and my_kwarg for some purpose
        ...
        await app(scope, receive, send)

    return my_middleware


app = Litestar(
    route_handlers=[...],
    middleware=[DefineMiddleware(middleware_factory, 1, my_kwarg="abc")],
)

The DefineMiddleware is a simple container - it takes a middleware callable as a first parameter, and then any positional arguments, followed by key word arguments. The middleware callable will be called with these values as well as the kwarg app as mentioned above.