Middleware

Lilya includes several middleware classes unique to the application but also allowing some other ways of designing them by using protocols.

Lilya middleware

The Lilya middleware is the classic already available way of declaring the middleware within an Lilya application.

from lilya.apps import Lilya
from lilya.middleware import DefineMiddleware
from lilya.middleware.httpsredirect import HTTPSRedirectMiddleware
from lilya.middleware.trustedhost import TrustedHostMiddleware

app = Lilya(
    routes=[...],
    middleware=[
        DefineMiddleware(
            TrustedHostMiddleware,
            allowed_hosts=["example.com", "*.example.com"],
        ),
        DefineMiddleware(HTTPSRedirectMiddleware),
    ],
)

Lilya protocols

Lilya protocols are not too different from the Lilya middleware. In fact, the name itself happens only because of the use of the python protocols which forces a certain structure to happen.

When designing a middleware, you can inherit and subclass the MiddlewareProtocol provided by Lilya.

from contextlib import AsyncExitStack
from typing import Optional

from lilya.protocols.middleware import MiddlewareProtocol
from lilya.types import ASGIApp, Receive, Scope, Send


class AsyncExitStackMiddleware(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp"):
        """AsyncExitStack Middleware class.

        Args:
            app: The 'next' ASGI app to call.
            config: The AsyncExitConfig instance.
        """
        super().__init__(app)
        self.app = app

    async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        if not AsyncExitStack:
            await self.app(scope, receive, send)  # pragma: no cover

        exception: Optional[Exception] = None
        async with AsyncExitStack() as stack:
            scope["lilya_astack"] = stack
            try:
                await self.app(scope, receive, send)
            except Exception as e:
                exception = e
                raise e
        if exception:
            raise exception

MiddlewareProtocol

For those coming from a more enforced typed language like Java or C#, a protocol is the python equivalent to an interface.

The MiddlewareProtocol is simply an interface to build middlewares for Lilya by enforcing the implementation of the __init__ and the async def __call__.

Enforcing this protocol also aligns with writing a Pure ASGI Middleware.

Quick sample

from typing import Any, Dict

from lilya.protocols.middleware import MiddlewareProtocol
from lilya.types import ASGIApp, Receive, Scope, Send


class SampleMiddleware(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp", **kwargs):
        """SampleMiddleware Middleware class.

        The `app` is always enforced.

        Args:
            app: The 'next' ASGI app to call.
            kwargs: Any arbitrarty data.
        """
        super().__init__(app)
        self.app = app
        self.kwargs = kwargs

    async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        """
        Implement the middleware logic here
        """
        ...


class AnotherSample(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp", **kwargs: Dict[str, Any]):
        super().__init__(app, **kwargs)
        self.app = app

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

Middleware and the application

Creating this type of middlewares will make sure the protocols are followed and therefore reducing development errors by removing common mistakes.

To add middlewares to the application is very simple.

Application levelAny other level

from typing import Any, Dict

from lilya.apps import Lilya
from lilya.middleware import DefineMiddleware
from lilya.protocols.middleware import MiddlewareProtocol
from lilya.types import ASGIApp, Receive, Scope, Send


class SampleMiddleware(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp", **kwargs):
        """SampleMiddleware Middleware class.

        The `app` is always enforced.

        Args:
            app: The 'next' ASGI app to call.
            kwargs: Any arbitrarty data.
        """
        super().__init__(app)
        self.app = app
        self.kwargs = kwargs

    async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        """
        Implement the middleware logic here
        """
        ...


class AnotherSample(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp", **kwargs: Dict[str, Any]):
        super().__init__(app, **kwargs)
        self.app = app

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


app = Lilya(
    routes=[...],
    middleware=[
        DefineMiddleware(SampleMiddleware),
        DefineMiddleware(AnotherSample),
    ],
)

from typing import Any, Dict

from lilya.apps import Lilya
from lilya.middleware import DefineMiddleware
from lilya.protocols.middleware import MiddlewareProtocol
from lilya.routing import Include, Path
from lilya.types import ASGIApp, Receive, Scope, Send


class SampleMiddleware(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp", **kwargs):
        """SampleMiddleware Middleware class.

        The `app` is always enforced.

        Args:
            app: The 'next' ASGI app to call.
            kwargs: Any arbitrarty data.
        """
        super().__init__(app)
        self.app = app
        self.kwargs = kwargs

    async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        """
        Implement the middleware logic here
        """
        ...


class AnotherSample(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp", **kwargs: Dict[str, Any]):
        super().__init__(app, **kwargs)
        self.app = app

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


class CustomMiddleware(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp", **kwargs: Dict[str, Any]):
        super().__init__(app, **kwargs)
        self.app = app

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


async def home():
    return "Hello world"


# Via Path
app = Lilya(
    routes=[Path("/", handler=home, middleware=[DefineMiddleware(AnotherSample)])],
    middleware=[DefineMiddleware(SampleMiddleware)],
)


# Via Include
app = Lilya(
    routes=[
        Include(
            "/",
            routes=[Path("/", handler=home, middleware=[DefineMiddleware(SampleMiddleware)])],
            middleware=[DefineMiddleware(CustomMiddleware)],
        )
    ],
    middleware=[DefineMiddleware(AnotherSample)],
)

Quick note

Info

The middleware is not limited to Lilya, ChildLilya, Include and Path. We simply choose Path as it looks simpler to read and understand.

Pure ASGI Middleware

Lilya follows the ASGI spec. This capability allows for the implementation of ASGI middleware using the ASGI interface directly. This involves creating a chain of ASGI applications that call into the next one. Notably, this approach mirrors the implementation of middleware classes shipped with Lilya.

Example of the most common approach

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


class MyMiddleware:
    def __init__(self, app: ASGIApp):
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        await self.app(scope, receive, send)

When implementing a Pure ASGI middleware, it is like implementing an ASGI application, the first parameter should always be an app and the __call__ should always return the app.

BaseAuthMiddleware

This is a very special middleware and helps with any authentication middleware that can be used within an Lilya application but like everything else, you can design your own.

BaseAuthMiddleware is also a protocol that simply enforces the implementation of the authenticate method and assigning the result object into a AuthResult and make it available on every request.

Example of a JWT middleware class

/src/middleware/jwt.py

from myapp.models import User
from myapp.security.jwt.token import Token
from saffier.exceptions import ObjectNotFound

from lilya._internal._connection import Connection
from lilya.exceptions import NotAuthorized
from lilya.middleware.authentication import AuthResult, BaseAuthMiddleware
from lilya.types import ASGIApp


class JWTAuthMiddleware(BaseAuthMiddleware):
    """
    An example how to integrate and design a JWT authentication
    middleware assuming a `myapp` in Lilya.
    """

    def __init__(
        self,
        app: ASGIApp,
        signing_key: str,
        algorithm: str,
        api_key_header: str,
    ):
        super().__init__(app)
        self.app = app
        self.signing_key = signing_key
        self.algorithm = algorithm
        self.api_key_header = api_key_header

    async def retrieve_user(self, user_id) -> User:
        try:
            return await User.get(pk=user_id)
        except ObjectNotFound:
            raise NotAuthorized()

    async def authenticate(self, request: Connection) -> AuthResult:
        token = request.headers.get(self.api_key_header)

        if not token:
            raise NotAuthorized("JWT token not found.")

        token = Token.decode(token=token, key=self.signing_key, algorithm=self.algorithm)

        user = await self.retrieve_user(token.sub)
        return AuthResult(user=user)

1. Import the BaseAuthMiddleware and AuthResult from lilya.middleware.authentication.
2. Implement the authenticate and assign the user result to the AuthResult.

Import the middleware into a Lilya application

From the application instanceFrom the settings

from lilya import Lilya
from lilya.middleware import DefineMiddleware
from .middleware.jwt import JWTAuthMiddleware


app = Lilya(routes=[...], middleware=[DefineMiddleware(JWTAuthMiddleware)])

from dataclasses import dataclass
from typing import List

from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from .middleware.jwt import JWTAuthMiddleware


@dataclass
class AppSettings(Settings):

    @property
    def middleware(self) -> List[DefineMiddleware]:
        return [
            DefineMiddleware(JWTAuthMiddleware)
        ]

# load the settings via LILYA_SETTINGS_MODULE=src.configs.live.AppSettings
app = Lilya(routes=[...])

Tip

To know more about loading the settings and the available properties, have a look at the settings docs.

Middleware and the settings

One of the advantages of Lilya is leveraging the settings to make the codebase tidy, clean and easy to maintain. As mentioned in the settings document, the middleware is one of the properties available to use to start a Lilya application.

src/configs/live.py

from __future__ import annotations

from dataclasses import dataclass

from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.compression import GZipMiddleware
from lilya.middleware.httpsredirect import HTTPSRedirectMiddleware


@dataclass
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        """
        All the middlewares to be added when the application starts.
        """
        return [
            DefineMiddleware(HTTPSRedirectMiddleware),
            DefineMiddleware(GZipMiddleware, minimum_size=500, compresslevel=9),
        ]

Start the application with the new settings

LILYA_SETTINGS_MODULE=configs.live.AppSettings uvicorn src:app

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [28720]
INFO:     Started server process [28722]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

Attention

If LILYA_SETTINGS_MODULE is not specified as the module to be loaded, Lilya will load the default settings but your middleware will not be initialized.

Important

If you need to specify parameters in your middleware then you will need to wrap it in a lilya.middleware.DefineMiddleware object to do it so. See GZipMiddleware example.

Available middlewares

• CSRFMiddleware - Handles with the CSRF.
• CORSMiddleware - Handles with the CORS.
• TrustedHostMiddleware - Handles with the CORS if a given allowed_hosts is populated.
• GZipMiddleware - Compression middleware gzip.
• HTTPSRedirectMiddleware - Middleware that handles HTTPS redirects for your application. Very useful to be used for production or production like environments.
• SessionMiddleware - Middleware that handles the sessions.
• WSGIMiddleware - Allows to connect WSGI applications and run them inside Lilya. A great example how to use it is available.
• XFrameOptionsMiddleware - Middleware that handles specifically against clickjacking.
• SecurityMiddleware - Provides several security enhancements to the request/response cycle and adds security headers to the response.

CSRFMiddleware

The default parameters used by the CSRFMiddleware implementation are restrictive by default and Lilya allows some ways of using this middleware depending of the taste.

from __future__ import annotations

from dataclasses import dataclass

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.csrf import CSRFMiddleware

routes = [...]

# Option one
middleware = [DefineMiddleware(CSRFMiddleware, secret="your-long-unique-secret")]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
@dataclass
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(CSRFMiddleware, secret="your-long-unique-secret"),
        ]

CORSMiddleware

The default parameters used by the CORSMiddleware implementation are restrictive by default and Lilya allows some ways of using this middleware depending of the taste.

from __future__ import annotations

from dataclasses import dataclass

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.cors import CORSMiddleware

routes = [...]

# Option one
middleware = [DefineMiddleware(CORSMiddleware, allow_origins=["*"])]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
@dataclass
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(CORSMiddleware, allow_origins=["*"]),
        ]

SessionMiddleware

Adds signed cookie-based HTTP sessions. Session information is readable but not modifiable.

from __future__ import annotations

from dataclasses import dataclass

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.sessions import SessionMiddleware

routes = [...]

# Option one
middleware = [DefineMiddleware(SessionMiddleware, secret_key=...)]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
@dataclass
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(SessionMiddleware, secret_key=...),
        ]

HTTPSRedirectMiddleware

Enforces that all incoming requests must either be https or wss. Any http os ws will be redirected to the secure schemes instead.

from __future__ import annotations

from dataclasses import dataclass

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.httpsredirect import HTTPSRedirectMiddleware

routes = [...]

# Option one
middleware = [DefineMiddleware(HTTPSRedirectMiddleware)]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
@dataclass
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(HTTPSRedirectMiddleware),
        ]

TrustedHostMiddleware

Enforces all requests to have a correct set Host header in order to protect against host header attacks.

from __future__ import annotations

from dataclasses import dataclass

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.trustedhost import TrustedHostMiddleware

routes = [...]

# Option one
middleware = [
    DefineMiddleware(TrustedHostMiddleware, allowed_hosts=["www.example.com", "*.example.com"])
]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
@dataclass
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(
                TrustedHostMiddleware, allowed_hosts=["www.example.com", "*.example.com"]
            ),
        ]

GZipMiddleware

It handles GZip responses for any request that includes "gzip" in the Accept-Encoding header.

from __future__ import annotations

from dataclasses import dataclass

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.compression import GZipMiddleware

routes = [...]

middleware = [DefineMiddleware(GZipMiddleware, minimum_size=1000)]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
@dataclass
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(GZipMiddleware, minimum_size=1000),
        ]

WSGIMiddleware

A middleware class in charge of converting a WSGI application into an ASGI one. There are some more examples in the WSGI Frameworks section.

from flask import Flask, make_response

from lilya.apps import Lilya
from lilya.middleware.wsgi import WSGIMiddleware
from lilya.routing import Include

flask = Flask(__name__)


@flask.route("/home")
def home():
    return make_response({"message": "Serving via flask"})


# Add the flask app into Lilya to be served by Lilya.
routes = [Include("/external", app=WSGIMiddleware(flask))]

app = Lilya(routes=routes)

The WSGIMiddleware also allows to pass the app as a string <dotted>.<path> and this can make it easier for code organisation.

Let us assume the previous example of the flask app was inside myapp/asgi_or_wsgi/apps. Like this:

from flask import Flask, make_response

flask = Flask(__name__)


@flask.route("/home")
def home():
    return make_response({"message": "Serving via flask"})

To call it inside the middleware is as simple as:

from lilya.apps import Lilya
from lilya.middleware.wsgi import WSGIMiddleware
from lilya.routing import Include

# Add the flask app into Lilya to be served by Lilya.
routes = [
    Include(
        "/external",
        app=WSGIMiddleware("myapp.asgi_or_wsgi.apps.flask"),
    ),
]

app = Lilya(routes=routes)

XFrameOptionsMiddleware

The clickjacking middleware that provides easy-to-use protection against clickjacking. This type of attack occurs when a malicious site tricks a user into clicking on a concealed element of another site which they have loaded in a hidden frame or iframe.

This middleware reads the value x_frame_options from the settings and defaults to DENY.

This also adds the X-Frame-Options to the response headers.

from __future__ import annotations

from dataclasses import dataclass

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.clickjacking import XFrameOptionsMiddleware

routes = [...]

# Option one
middleware = [DefineMiddleware(XFrameOptionsMiddleware)]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
@dataclass
class AppSettings(Settings):
    x_frame_options: str = "SAMEORIGIN"

    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(XFrameOptionsMiddleware),
        ]

SecurityMiddleware

Provides several security enhancements to the request/response cycle and adds security headers to the response.

from __future__ import annotations

from dataclasses import dataclass

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.security import SecurityMiddleware

routes = [...]

content_policy_dict = {
    "default-src": "'self'",
    "img-src": [
        "*",
        "data:",
    ],
    "connect-src": "'self'",
    "script-src": "'self'",
    "style-src": ["'self'", "'unsafe-inline'"],
    "script-src-elem": [
        "https://unpkg.com/@stoplight/elements/web-components.min.jss",
    ],
    "style-src-elem": [
        "https://unpkg.com/@stoplight/elements/styles.min.css",
    ],
}

# Option one
middleware = [DefineMiddleware(SecurityMiddleware, content_policy=content_policy_dict)]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
@dataclass
class AppSettings(Settings):

    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(SecurityMiddleware, content_policy=content_policy_dict),
        ]

ClientIPMiddleware

Parses the client ip and add it in the request scope at two places: headers ("x-real-ip") and the request scope directly ("real-clientip").

from __future__ import annotations

from dataclasses import dataclass

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.clientip import ClientIPMiddleware

routes = [...]

# Only trust unix (no client ip is set) (default)
trusted_proxies = ["unix"]
# No forwarded ip headers will be evaluated. Only the direct ip is accepted
trusted_proxies = []
# trust all client ips to provide forwarded headers
trusted_proxies = ["*"]


# Option one
middleware = [DefineMiddleware(ClientIPMiddleware, trusted_proxies=trusted_proxies)]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
@dataclass
class AppSettings(Settings):

    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(ClientIPMiddleware, trusted_proxies=trusted_proxies),
        ]

There are two special "ip"s: "*" and "unix"

The first one is a match all and implies all proxies are trustworthy, the second one applies when a unix socket is used or no client ip address was found.

Note

If you don't want to use the middleware you can use: get_ip from lilya.clientip directly.

Note

It is currently not possible to simulate a client ip address in lilyas TestClient. So you may want to use the Forwarded header and trust "unix" for tests.

Other middlewares

You can build your own middlewares as explained above but also reuse middlewares directly for any other ASGI application if you wish. If the middlewares follow the pure asgi then the middlewares are 100% compatible.

RateLimitMiddleware

A ASGI Middleware to rate limit and highly customizable.

CorrelationIdMiddleware

A middleware class for reading/generating request IDs and attaching them to application logs.

Tip

For Lilya apps, just substitute FastAPI with Lilya in the examples given or implement in the way Lilya shows in this document.

TimingMiddleware

ASGI middleware to record and emit timing metrics (to something like statsd).

Important points

1. Lilya supports Lilya middleware (MiddlewareProtocol).
2. A MiddlewareProtocol is simply an interface that enforces __init__ and async __call__ to be implemented.
3. app is required parameter from any class inheriting from the MiddlewareProtocol.
4. Pure ASGI Middleware is encouraged and the MiddlewareProtocol enforces that.
5. Middleware classes can be added to any layer of the application
6. All authentication middlewares must inherit from the BaseAuthMiddleware.
7. You can load the application middleware in different ways.