Using Middleware#

A middleware in Starlite 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 starlite.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 Starlite instance, where it will be called on every request:

from starlite.types import ASGIApp, Scope, Receive, Send
from starlite import Starlite


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 = Starlite(route_handlers=[...], middleware=[middleware_factory])

In the above example, Starlite will call the middleware_factory function and pass to it app. It’s important to understand that this kwarg does not designate the Starlite 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 Starlite’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 starlite import Controller, MiddlewareProtocol, Router, Starlite, State, get

if TYPE_CHECKING:
    from starlite.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:
            starlite_app = scope["app"]
            starlite_app.state.setdefault("middleware_calls", [])
            starlite_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"]  # type: ignore


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

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


# run: /router/controller/handler
from typing import TYPE_CHECKING

from starlite import Controller, MiddlewareProtocol, Router, Starlite, State, get

if TYPE_CHECKING:
    from starlite.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:
            starlite_app = scope["app"]
            starlite_app.state.setdefault("middleware_calls", [])
            starlite_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"]  # type: ignore


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

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


# run: /router/controller/handler

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 Starlite’s ASGI router:

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