Middleware

O Lilya inclui vários middlewares exclusivos da aplicação, mas também permite algumas maneiras de os criar utilizando protocolos.

Middleware Lilya

O middleware Lilya é a maneira clássica já disponível de declarar o middleware dentro de uma aplicação Lilya.

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

app = Lilya(
    routes=[...],
    middleware=[
        DefineMiddleware(
            TrustedHostMiddleware,
            allowed_hosts=["example.com", "*.example.com"],
        ),
        # you can also use import strings
        DefineMiddleware("lilya.middleware.httpsredirect.HTTPSRedirectMiddleware"),
    ],
)

Protocolos Lilya

Os protocolos Lilya não são muito diferentes do middleware Lilya. Na verdade, o nome em si só existe por causa do uso dos protocolos Python, que forçam uma certa estrutura.

Ao criar um middleware, pode herdar o MiddlewareProtocol fornecido pelo 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

Para aqueles que estão habituados a linguagens de programação com forte enfase no static typing, como Java ou C#, um protocolo é o equivalente em Python a uma interface.

O MiddlewareProtocol é simplesmente uma interface para construir middlewares para o Lilya, forçando a implementação dos métodos __init__ e async def __call__.

O uso desse protocolo também está alinhado com a criação de um Middleware ASGI Puro.

Exemplo rápido

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 e a aplicação

A criação deste tipo de middleware garantirá que os protocolos sejam seguidos, reduzindo assim erros de desenvolvimento ao remover erros comuns.

Para adicionar middlewares à aplicação é muito simples.

Nível da aplicaçãoQualquer outro nível

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)],
)

Nota rápida

Info

O middleware não se limita ao Lilya, ChildLilya, Include e Path. Apenas foi escolhido o Path porque é mais simples de ler e perceber.

Middleware ASGI Puro

O Lilya segue a especificação do ASGI. Essa capacidade permite a implementação de middlewares ASGI utilizando a interface ASGI diretamente. Isso envolve a criação de uma cadeia de aplicações ASGI que chamam o seguinte.

A abordagem espelha a implementação das classes middleware fornecidas pelo Lilya.

Exemplo da abordagem mais comum

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)

Ao implementar um middleware ASGI Puro, é como implementar uma aplicação ASGI, o primeiro parâmetro deve ser sempre uma aplicação e o método __call__ deve retornar sempre a aplicação.

BaseAuthMiddleware

Este é um middleware muito especial e ajuda qualquer middleware relacionado com autenticação que pode ser usado numa aplicação Lilya, mas, como tudo, também pode criar o seu próprio e ignorar isto.

BaseAuthMiddleware é também um protocolo que simplesmente força a implementação do método authenticate e atribui o objecto de resultado a um AuthResult para torná-lo disponível em cada pedido.

Exemplo de uma classe de middleware JWT

/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.authentication import AuthResult
from lilya.middleware.authentication import 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 (None, user)

1. Importe o BaseAuthMiddleware de lilya.middleware.authentication.
2. Implemente o método authenticate e atribua o resultado user ao tuple[AuthCredentials, UserInterface] (AuthResult).

Importe o middleware numa aplicação Lilya

A partir da instância da aplicaçãoA partir das definições

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)
        ]

# carregue as definições via LILYA_SETTINGS_MODULE=src.configs.live.AppSettings
app = Lilya(routes=[...])

Tip

Para saber mais sobre como carregar as definições e as propriedades disponíveis, consulte a documentação das definições.

Middleware e as definições

Uma das vantagens do Lilya é aproveitar as definições para tornar o código organizado, limpo e fácil de manter. Conforme mencionado no documento de definições, o middleware é uma das propriedades disponíveis para iniciar uma aplicação Lilya.

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),
        ]

Inicie a aplicação com as novas definições

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.

Warning

Se o LILYA_SETTINGS_MODULE não for especificado como o módulo a ser carregado, o Lilya carregará as definições padrão, mas o seu middleware não será inicializado.

Importante

Se precisar especificar parâmetros no seu middleware, será necessário encapsulá-lo num objecto lilya.middleware.DefineMiddleware. Veja o exemplo do GZipMiddleware aqui.

Middlewares disponíveis

• CSRFMiddleware - Lida com CSRF.
• CORSMiddleware - Lida com CORS.
• TrustedHostMiddleware - Lida com CORS se um allowed_hosts específico estiver definido.
• GZipMiddleware - Middleware de compressão gzip.
• HTTPSRedirectMiddleware - Middleware que lida com redirecionamentos HTTPS para a sua aplicação. Muito útil para uso em ambientes de produção ou semelhantes a produção.
• SessionMiddleware - Middleware que lida com sessões.
• WSGIMiddleware - Permite ligar aplicações WSGI e executá-los dentro do Lilya. Um ótimo exemplo de como usá-lo está disponível.
• XFrameOptionsMiddleware - Middleware que lida especificamente contra clickjacking.
• SecurityMiddleware - Fornece várias melhorias de segurança ao ciclo de pedido/resposta e adiciona cabeçalhos de segurança à resposta.

CSRFMiddleware

Os parâmetros padrão usados na implementação do CSRFMiddleware são restritivos por defeito e o Lilya permite algumas maneiras de usar esse middleware, dependendo das preferências.

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

Os parâmetros padrão usados na implementação do CORSMiddleware são restritivos por defeito e o Lilya permite algumas maneiras de usar esse middleware, dependendo das preferências.

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

Adiciona sessões HTTP baseadas em cookies assinados. As informações da sessão são legíveis, mas não modificáveis.

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
from lilya.requests import Request
from lilya.routing import Path


async def set_session(request: Request) -> dict:
    # this creates/updates the session
    if not request.scope["session"]:
        request.scope["session"] = {"id": 1, "seen_page_list": []}
    else:
        # this updates the session
        request.scope["session"]["seen_page_list"].append("homepage")
    return request.scope["session"]


async def delete_session(request: Request) -> dict:
    old_session = request.scope["session"]
    request.scope["session"] = None
    return old_session

routes = [Path("/set", set_session), Path("/delete", delete_session)]

# 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

Garante que todos os pedidos recebidas devem ser https ou wss. Qualquer pedido http ou ws será redirecionado para o formato seguro correspondente.

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

Exige que todos os pedidos tenham um cabeçalho Host corretamente definido para proteção contra ataques host 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.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

Lida com respostas GZip para qualquer pedido que inclua "gzip" no cabeçalho Accept-Encoding.

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

Uma classe de middleware responsável por converter uma aplicação WSGI numa aplicação ASGI. Existem mais exemplos na secção Frameworks WSGI.

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)

O WSGIMiddleware também permite passar a app como uma string <dotted>.<path>, o que pode facilitar a organização do código.

Vamos supor que o exemplo anterior da aplicação flask estivesse dentro da myapp/asgi_or_wsgi/apps. Ficaria desta forma:

from flask import Flask, make_response

flask = Flask(__name__)


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

Para chamá-lo dentro do middleware é tão simples quanto isto:

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

O middleware de clickjacking fornece proteção fácil de usar contra ataques de clickjacking. Este tipo de ataque ocorre quando um site malicioso engana um utilizador para clicar num elemento oculto de outro site que eles carregaram num iframe oculto.

Este middleware lê o valor x_frame_options das configurações e tem como valor padrão DENY.

Ele também adiciona o cabeçalho X-Frame-Options às respostas.

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

Fornece várias melhorias de segurança no ciclo de pedido/resposta e adiciona cabeçalhos de segurança à resposta.

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),
        ]

Outros middlewares

Pode desenhar os seus próprios middlewares conforme explicado acima, mas também reutilizar middlewares diretamente para qualquer outra aplicação ASGI, se assim o desejar. Se os middlewares seguirem a abordagem do ASGI puro, eles serão 100% compatíveis.

RateLimitMiddleware

Um Middleware ASGI para limitar a taxa de pedidos e altamente personalizável.

CorrelationIdMiddleware

Uma classe de middleware para ler/gerar IDs de pedido e anexá-los aos logs da aplicação.

Tip

Para aplicações Lilya, substitua FastAPI por Lilya nos exemplos fornecidos ou implemente da maneira mostrada neste documento.

TimingMiddleware

Middleware ASGI para registrar e emitir métricas de tempo (para algo como statsd).

Pontos importantes

1. O Lilya oferece suporte ao middleware Lilya (MiddlewareProtocol).
2. MiddlewareProtocol é simplesmente uma interface que exige a implementação do __init__ e async __call__.
3. app é um parâmetro obrigatório para qualquer classe que herda do MiddlewareProtocol.
4. É encorajado o uso de Middleware ASGI Puro e o MiddlewareProtocol exige isso.
5. As classes middleware podem ser adicionadas em qualquer camada da aplicação.
6. Todos os middlewares de autenticação devem herdar do BaseAuthMiddleware.
7. Pode carregar o middleware da aplicação de diferentes maneiras.