Middleware

Esmerald includes several middleware classes unique to the application but also allowing some other ways of designing them by using protocols. Inspired by other great frameworks, Esmerald has a similar approach for the middleware protocol. Let's be honest, it is not that we can reinvent the wheel on something already working out of the box.

There are two ways of designing the middleware for Esmerald. Lilya middleware and Esmerald protocols as both work quite well together.

Lilya middleware

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

Tip

You can create a middleware like Lilya and add it into the application. To understand how to build them, Lilya has some great documentation here.

from esmerald import Esmerald
from esmerald.middleware import HTTPSRedirectMiddleware, TrustedHostMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

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

The example above is for illustration purposes only as those middlewares are already in place based on specific configurations passed into the application instance. Have a look at CORSConfig, CSRFConfig, SessionConfig to understand how to use them and automatically enable the built-in middlewares.

Esmerald protocols

Esmerald 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 and since Esmerald likes configurations as much as possible, using a protocol helps enforcing that and allows a better design.

from typing import Optional

from esmerald.concurrency import AsyncExitStack
from esmerald.config import AsyncExitConfig
from esmerald.protocols.middleware import MiddlewareProtocol
from esmerald.types import ASGIApp, Receive, Scope, Send


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

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

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

        exception: Optional[Exception] = None
        async with AsyncExitStack() as stack:
            scope[self.config.context_name] = 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 Esmerald by enforcing the implementation of the __init__ and the async def __call__.

In the case of Esmerald configurations, a config parameter is declared and passed in the __init__ but this is not enforced on a protocol level but on a subclass level, the middleware itself.

Enforcing this protocol also aligns with writing pure asgi middlewares.

Note

MiddlewareProtocol does not enforce config parameters but enforces the app parameter as this will make sure it will also work with Lilya as well as used as standard.

Quick sample

from typing import Any, Dict

from esmerald.protocols.middleware import MiddlewareProtocol
from esmerald.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)

MiddlewareProtocol 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 esmerald import Esmerald
from esmerald.protocols.middleware import MiddlewareProtocol
from esmerald.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 = Esmerald(routes=[...], middleware=[SampleMiddleware, AnotherSample])

from typing import Any, Dict

from esmerald import Esmerald, Gateway, Include, get
from esmerald.protocols.middleware import MiddlewareProtocol
from esmerald.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: ...


@get()
async def home() -> str:
    return "Hello world"


# Via Gateway

app = Esmerald(
    routes=[Gateway(handler=get, middleware=[AnotherSample])],
    middleware=[SampleMiddleware],
)


# Via Include

app = Esmerald(
    routes=[
        Include(
            routes=[Gateway(handler=get, middleware=[SampleMiddleware])],
            middleware=[CustomMiddleware],
        )
    ],
    middleware=[AnotherSample],
)

Quick note

Info

The middleware is not limited to Esmerald, ChildEsmerald, Include and Gateway. They also work with WebSocketGateway and inside every get, post, put, patch, delete and route as well as websocket. We simply choose Gateway as it looks simpler to read and understand.

Writing ASGI middlewares

Esmerald since follows the ASGI practices and uses Lilya underneath a good way of understand what can be done with middleware and how to write some of them, Lilya also goes through with a lot of detail.

BaseAuthMiddleware

This is a very special middleware and it is the core for every authentication middleware that is used within an Esmerald application.

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.

API Reference

Check out the API Reference for BasseAuthMiddleware for more details.

Example of a JWT middleware class

/src/middleware/jwt.py

from esmerald.config.jwt import JWTConfig
from esmerald.contrib.auth.saffier.base_user import User
from esmerald.exceptions import NotAuthorized
from esmerald.middleware.authentication import AuthResult, BaseAuthMiddleware
from esmerald.security.jwt.token import Token
from lilya._internal._connection import Connection
from lilya.types import ASGIApp
from saffier.exceptions import ObjectNotFound


class JWTAuthMiddleware(BaseAuthMiddleware):
    def __init__(self, app: "ASGIApp", config: "JWTConfig"):
        super().__init__(app)
        self.app = app
        self.config = config

    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.config.api_key_header)

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

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

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

1. Import the BaseAuthMiddleware and AuthResult from esmerald.middleware.authentication.
2. Import JWTConfig to pass some specific and unique JWT configations into the middleware.
3. Implement the authenticate and assign the user result to the AuthResult.

Info

We use Saffier for this example because Esmerald supports S and contains functionalities linked with that support (like the User table) but Esmerald is not dependent of ANY specific ORM which means that you are free to use whatever you prefer.

Import the middleware into an Esmerald application

From the application instanceFrom the settings

from esmerald import Esmerald
from .middleware.jwt import JWTAuthMiddleware


app = Esmerald(routes=[...], middleware=[JWTAuthMiddleware])

from typing import List

from esmerald import EsmeraldAPISettings
from esmerald.types import Middleware
from .middleware.jwt import JWTAuthMiddleware


class AppSettings(EsmeraldAPISettings):

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

# load the settings via ESMERALD_SETTINGS_MODULE=src.configs.live.AppSettings
app = Esmerald(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 Esmerald 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 an Esmerald application.

src/configs/live.py

from typing import List

from esmerald import EsmeraldAPISettings
from esmerald.middleware import GZipMiddleware, HTTPSRedirectMiddleware
from esmerald.types import Middleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware


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

Start the application with the new settings

MacOS & LinuxWindows

ESMERALD_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.

$env:ESMERALD_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 ESMERALD_SETTINGS_MODULE is not specified as the module to be loaded, Esmerald 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.

If no parameters are needed, then you can simply pass the middleware class directly and Esmerald will take care of the rest.

Available middlewares

There are some available middlewares that are also available from Lilya.

• CSRFMiddleware - Handles with the CSRF and there is a built-in how to enable.
• CORSMiddleware - Handles with the CORS and there is a built-in how to enable.
• TrustedHostMiddleware - Handles with the CORS if a given allowed_hosts is populated, the built-in explains how to use it.
• GZipMiddleware - Same middleware as the one from Lilya.
• HTTPSRedirectMiddleware - Middleware that handles HTTPS redirects for your application. Very useful to be used for production or production like environments.
• RequestSettingsMiddleware - The middleware that exposes the application settings in the request.
• SessionMiddleware - Same middleware as the one from Lilya.
• WSGIMiddleware - Allows to connect WSGI applications and run them inside Esmerald. A great example how to use it is available.

CSRFMiddleware

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

from esmerald import Esmerald, EsmeraldAPISettings
from esmerald.config import CSRFConfig
from esmerald.middleware import CSRFMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

routes = [...]

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

app = Esmerald(routes=routes, middleware=middleware)


# Option two - Activating the built-in middleware using the config.
csrf_config = CSRFConfig(secret="your-long-unique-secret")

app = Esmerald(routes=routes, csrf_config=csrf_config)


# Option three - Using the settings module
# Running the application with your custom settings -> ESMERALD_SETTINGS_MODULE
class AppSettings(EsmeraldAPISettings):
    @property
    def csrf_config(self) -> CSRFConfig:
        return CSRFConfig(allow_origins=["*"])

CORSMiddleware

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

from esmerald import Esmerald, EsmeraldAPISettings
from esmerald.config import CORSConfig
from esmerald.middleware import CORSMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

routes = [...]

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

app = Esmerald(routes=routes, middleware=middleware)


# Option two - Activating the built-in middleware using the config.
cors_config = CORSConfig(allow_origins=["*"])

app = Esmerald(routes=routes, cors_config=cors_config)


# Option three - Using the settings module
# Running the application with your custom settings -> ESMERALD_SETTINGS_MODULE
class AppSettings(EsmeraldAPISettings):
    @property
    def cors_config(self) -> CORSConfig:
        return CORSConfig(allow_origins=["*"])

RequestSettingsMiddleware

Exposes your Esmerald application settings in the request. This can be particulary useful to access the main settings module in any part of the application, inclusively ChildEsmerald.

This middleware has settings as optional parameter. If none is provided it will default to the internal settings.

RequestSettingsMiddleware adds two types of settings to the request, the global_settings where is the global Esmerald settings and the app_settings which corresponds to the settings_module, if any, passed to the Esmerald or ChildEsmerald instance.

from esmerald import Esmerald
from esmerald.middleware import RequestSettingsMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

middleware = [LilyaMiddleware(RequestSettingsMiddleware)]

app = Esmerald(routes=[...], middleware=middleware)

SessionMiddleware

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

from esmerald import Esmerald, EsmeraldAPISettings
from esmerald.config import SessionConfig
from esmerald.middleware import SessionMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

routes = [...]

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

app = Esmerald(routes=routes, middleware=middleware)


# Option two - Activating the built-in middleware using the config.
session_config = SessionConfig(secret_key=...)

app = Esmerald(routes=routes, session_config=session_config)


# Option three - Using the settings module
# Running the application with your custom settings -> ESMERALD_SETTINGS_MODULE
class AppSettings(EsmeraldAPISettings):
    @property
    def session_config(self) -> SessionConfig:
        return SessionConfig(secret_key=...)

HTTPSRedirectMiddleware

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

from typing import List

from esmerald import Esmerald, EsmeraldAPISettings
from esmerald.middleware import HTTPSRedirectMiddleware
from esmerald.types import Middleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

routes = [...]

# Option one
middleware = [LilyaMiddleware(HTTPSRedirectMiddleware)]

app = Esmerald(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> ESMERALD_SETTINGS_MODULE
class AppSettings(EsmeraldAPISettings):
    @property
    def middleware(self) -> List["Middleware"]:
        # There is no need to wrap in a LilyaMiddleware here.
        # Esmerald automatically will do it once the application is up and running.
        return [HTTPSRedirectMiddleware]

TrustedHostMiddleware

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

from typing import List

from esmerald import Esmerald, EsmeraldAPISettings
from esmerald.middleware import TrustedHostMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

routes = [...]

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

app = Esmerald(routes=routes, middleware=middleware)


# Option two - Activating the built-in middleware using the config.
allowed_hosts = ["www.example.com", "*.example.com"]

app = Esmerald(routes=routes, allowed_hosts=allowed_hosts)


# Option three - Using the settings module
# Running the application with your custom settings -> ESMERALD_SETTINGS_MODULE
class AppSettings(EsmeraldAPISettings):
    allowed_hosts: List[str] = ["www.example.com", "*.example.com"]

GZipMiddleware

Like Lilya, it handles GZip responses for any request that includes "gzip" in the Accept-Encoding header.

from esmerald import Esmerald
from esmerald.middleware import GZipMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

routes = [...]

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

app = Esmerald(routes=routes, middleware=middleware)

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 esmerald import Esmerald, Include
from esmerald.middleware.wsgi import WSGIMiddleware

flask = Flask(__name__)


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


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

app = Esmerald(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 typing import List

from esmerald import Esmerald, EsmeraldAPISettings
from esmerald.middleware.clickjacking import XFrameOptionsMiddleware
from lilya.middleware import DefineMiddleware

routes = [...]

# Option one
middleware = [DefineMiddleware(XFrameOptionsMiddleware)]

app = Esmerald(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> ESMERALDS_SETTINGS_MODULE
class AppSettings(EsmeraldAPISettings):
    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 typing import List

from esmerald import Esmerald, EsmeraldAPISettings
from esmerald.middleware.security import SecurityMiddleware
from lilya.middleware import DefineMiddleware

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 = Esmerald(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> ESMERALD_SETTINGS_MODULE
class AppSettings(EsmeraldAPISettings):

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

Other middlewares

You can build your own middlewares as explained above but also reuse middlewares directly for Lilya if you wish. The middlewares are 100% compatible.

Although some of the middlewares might mention Lilya or other ASGI framework, they are 100% compatible with Esmerald as well.

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 Esmerald apps, just substitute FastAPI with Esmerald in the examples given or implement in the way Esmerald shows in this document.

TimingMiddleware

ASGI middleware to record and emit timing metrics (to something like statsd). This integration works using EsmeraldTimming.

Important points

1. Esmerald 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.