Middleware

Ravyn 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, Ravyn 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 Ravyn. Lilya middleware and Ravyn protocols as both work quite well together.

Lilya middleware

The Lilya middleware is the classic already available way of declaring the middleware within an Ravyn 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 ravyn import Ravyn
from ravyn.middleware import HTTPSRedirectMiddleware, TrustedHostMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

app = Ravyn(
    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.

Ravyn protocols

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

from typing import Optional

from ravyn.utils.concurrency import AsyncExitStack
from ravyn.core.config import AsyncExitConfig
from ravyn.core.protocols.middleware import MiddlewareProtocol
from ravyn.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 Ravyn by enforcing the implementation of the __init__ and the async def __call__.

In the case of Ravyn 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 ravyn.core.protocols.middleware import MiddlewareProtocol
from ravyn.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 ravyn import Ravyn
from ravyn.core.protocols.middleware import MiddlewareProtocol
from ravyn.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 = Ravyn(routes=[...], middleware=[SampleMiddleware, AnotherSample])

from typing import Any, Dict

from ravyn import Ravyn, Gateway, Include, get
from ravyn.core.protocols.middleware import MiddlewareProtocol
from ravyn.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 = Ravyn(
    routes=[Gateway(handler=get, middleware=[AnotherSample])],
    middleware=[SampleMiddleware],
)


# Via Include

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

Quick note

Info

The middleware is not limited to Ravyn, ChildRavyn, 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

Ravyn 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 Ravyn 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 ravyn.core.config.jwt import JWTConfig
from ravyn.contrib.auth.edgy.base_user import User
from ravyn.exceptions import NotAuthorized
from ravyn.middleware.authentication import AuthResult, BaseAuthMiddleware
from ravyn.security.jwt.token import Token
from lilya._internal._connection import Connection
from lilya.types import ASGIApp
from edgy.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 ravyn.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 Edgy for this example because Ravyn supports S and contains functionalities linked with that support (like the User table) but Ravyn is not dependent of ANY specific ORM which means that you are free to use whatever you prefer.

Import the middleware into an Ravyn application

From the application instanceFrom the settings

from ravyn import Ravyn
from .middleware.jwt import JWTAuthMiddleware


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

from typing import List

from ravyn import RavynSettings
from ravyn.types import Middleware
from .middleware.jwt import JWTAuthMiddleware


class AppSettings(RavynSettings):

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

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

src/configs/live.py

from typing import List

from ravyn import RavynSettings
from ravyn.middleware import GZipMiddleware, HTTPSRedirectMiddleware
from ravyn.types import Middleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware


class AppSettings(RavynSettings):
    @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

RAVYN_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:RAVYN_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 RAVYN_SETTINGS_MODULE is not specified as the module to be loaded, Ravyn 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 Ravyn 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 Ravyn. A great example how to use it is available.

CSRFMiddleware

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

from ravyn import Ravyn, RavynSettings
from ravyn.core.config import CSRFConfig
from ravyn.middleware import CSRFMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

routes = [...]

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

app = Ravyn(routes=routes, middleware=middleware)


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

app = Ravyn(routes=routes, csrf_config=csrf_config)


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

CORSMiddleware

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

from ravyn import Ravyn, RavynSettings
from ravyn.core.config import CORSConfig
from ravyn.middleware import CORSMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

routes = [...]

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

app = Ravyn(routes=routes, middleware=middleware)


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

app = Ravyn(routes=routes, cors_config=cors_config)


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

RequestSettingsMiddleware

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

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 Ravyn settings and the app_settings which corresponds to the settings_module, if any, passed to the Ravyn or ChildRavyn instance.

from ravyn import Ravyn
from ravyn.middleware import RequestSettingsMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

middleware = [LilyaMiddleware(RequestSettingsMiddleware)]

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

SessionMiddleware

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

from ravyn import Ravyn, RavynSettings
from ravyn.core.config import SessionConfig
from ravyn.middleware.sessions import SessionMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

routes = [...]

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

app = Ravyn(routes=routes, middleware=middleware)


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

app = Ravyn(routes=routes, session_config=session_config)


# Option three - Using the settings module
# Running the application with your custom settings -> RAVYN_SETTINGS_MODULE
class AppSettings(RavynSettings):
    @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 ravyn import Ravyn, RavynSettings
from ravyn.middleware import HTTPSRedirectMiddleware
from ravyn.types import Middleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

routes = [...]

# Option one
middleware = [LilyaMiddleware(HTTPSRedirectMiddleware)]

app = Ravyn(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> RAVYN_SETTINGS_MODULE
class AppSettings(RavynSettings):
    @property
    def middleware(self) -> List["Middleware"]:
        # There is no need to wrap in a LilyaMiddleware here.
        # Ravyn 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 ravyn import Ravyn, RavynSettings
from ravyn.middleware import TrustedHostMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

routes = [...]

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

app = Ravyn(routes=routes, middleware=middleware)


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

app = Ravyn(routes=routes, allowed_hosts=allowed_hosts)


# Option three - Using the settings module
# Running the application with your custom settings -> RAVYN_SETTINGS_MODULE
class AppSettings(RavynSettings):
    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 ravyn import Ravyn
from ravyn.middleware import GZipMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware

routes = [...]

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

app = Ravyn(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 ravyn import Ravyn, Include
from ravyn.middleware.wsgi import WSGIMiddleware

flask = Flask(__name__)


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


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

app = Ravyn(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 ravyn import Ravyn, RavynSettings
from ravyn.middleware.clickjacking import XFrameOptionsMiddleware
from lilya.middleware import DefineMiddleware

routes = [...]

# Option one
middleware = [DefineMiddleware(XFrameOptionsMiddleware)]

app = Ravyn(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> ESMERALDS_SETTINGS_MODULE
class AppSettings(RavynSettings):
    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 ravyn import Ravyn, RavynSettings
from ravyn.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 = Ravyn(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> RAVYN_SETTINGS_MODULE
class AppSettings(RavynSettings):
    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 Ravyn 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 Ravyn apps, just substitute FastAPI with Ravyn in the examples given or implement in the way Ravyn shows in this document.

Important points

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