Using Middleware#

A middleware in Litestar is any callable that receives at least one kwarg called app and returns an ASGIApp. An ASGIApp is nothing but an async function that receives the ASGI primitives scope , receive and send , and either calls the next ASGIApp or returns a response / handles the websocket connection.

For example, the following function can be used as a middleware because it receives the app kwarg and returns an ASGIApp:

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

We can then pass this middleware to the Litestar instance, where it will be called on every request:

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


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


app = Litestar(route_handlers=[...], middleware=[middleware_factory])

In the above example, Litestar will call the middleware_factory function and pass to it app. It’s important to understand that this kwarg does not designate the Litestar application but rather the next ASGIApp in the stack. It will then insert the returned my_middleware function into the stack of every route in the application - because we declared it on the application level.

Layered architecture

Middlewares are part of Litestar’s layered architecture* which means you can set them on every layer of the application.

You can read more about this here: Layered architecture

Middleware Call Order#

Since it’s also possible to define multiple middlewares on every layer, the call order for middlewares will be top to bottom and left to right. This means for each layer, the middlewares will be called in the order they have been passed, while the layers will be traversed in the usual order:

        flowchart LR
    Application --> Router --> Controller --> Handler
    
from typing import TYPE_CHECKING, List, Type

from litestar import Controller, Litestar, Router, get
from litestar.datastructures import State
from litestar.middleware import MiddlewareProtocol

if TYPE_CHECKING:
    from litestar.types import ASGIApp, Receive, Scope, Send


def create_test_middleware(middleware_id: int) -> Type[MiddlewareProtocol]:
    class TestMiddleware(MiddlewareProtocol):
        def __init__(self, app: "ASGIApp") -> None:
            self.app = app

        async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
            litestar_app = scope["app"]
            litestar_app.state.setdefault("middleware_calls", [])
            litestar_app.state["middleware_calls"].append(middleware_id)
            await self.app(scope, receive, send)

    return TestMiddleware


class MyController(Controller):
    path = "/controller"
    middleware = [create_test_middleware(4), create_test_middleware(5)]

    @get(
        "/handler",
        middleware=[create_test_middleware(6), create_test_middleware(7)],
    )
    async def my_handler(self, state: State) -> List[int]:
        return state["middleware_calls"]


router = Router(
    path="/router",
    route_handlers=[MyController],
    middleware=[create_test_middleware(2), create_test_middleware(3)],
)

app = Litestar(
    route_handlers=[router],
    middleware=[create_test_middleware(0), create_test_middleware(1)],
)

from typing import TYPE_CHECKING

from litestar import Controller, Litestar, Router, get
from litestar.datastructures import State
from litestar.middleware import MiddlewareProtocol

if TYPE_CHECKING:
    from litestar.types import ASGIApp, Receive, Scope, Send


def create_test_middleware(middleware_id: int) -> type[MiddlewareProtocol]:
    class TestMiddleware(MiddlewareProtocol):
        def __init__(self, app: "ASGIApp") -> None:
            self.app = app

        async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
            litestar_app = scope["app"]
            litestar_app.state.setdefault("middleware_calls", [])
            litestar_app.state["middleware_calls"].append(middleware_id)
            await self.app(scope, receive, send)

    return TestMiddleware


class MyController(Controller):
    path = "/controller"
    middleware = [create_test_middleware(4), create_test_middleware(5)]

    @get(
        "/handler",
        middleware=[create_test_middleware(6), create_test_middleware(7)],
    )
    async def my_handler(self, state: State) -> list[int]:
        return state["middleware_calls"]


router = Router(
    path="/router",
    route_handlers=[MyController],
    middleware=[create_test_middleware(2), create_test_middleware(3)],
)

app = Litestar(
    route_handlers=[router],
    middleware=[create_test_middleware(0), create_test_middleware(1)],
)

Run it

> curl http://127.0.0.1:8000/router/controller/handler
[0,1,2,3,4,5,6,7]

Middlewares and Exceptions#

When an exception is raised by a route handler or a dependency it will be transformed into a response by an exception handler. This response will follow the normal “flow” of the application and therefore, middlewares are still applied to it.

As with any good rule, there are exceptions to it. In this case they are two exceptions raised by Litestar’s ASGI router:

They are raised before the middleware stack is called and will only be handled by exception handlers defined on the Litestar instance itself. If you wish to modify error responses generated from these exception, you will have to use an application layer exception handler.