Перейти к содержанию

Middleware

Warning

The current page still doesn't have a translation for this language.

But you can help translating it: Contributing.

Middleware processes every request and response in your Ravyn application. Use it for logging, authentication, CORS, compression, and more. Ravyn supports both Lilya-style middleware and protocol-based middleware for maximum flexibility.

What You'll Learn

  • What middleware is and when to use it
  • Creating Lilya-style middleware
  • Using Ravyn's MiddlewareProtocol
  • Built-in middleware (CORS, CSRF, Sessions, etc.)
  • Adding middleware at different application levels
  • Authentication middleware patterns

Quick Start

from ravyn import Ravyn, get
from lilya.middleware import DefineMiddleware

# Simple logging middleware
class LoggingMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        print(f"Request: {scope['method']} {scope['path']}")
        await self.app(scope, receive, send)

@get("/")
def homepage() -> dict:
    return {"message": "Hello"}

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

Practical Middleware Patterns

This section covers real-world middleware patterns you can use in your Ravyn applications.

Request Logging Middleware

Logging middleware is essential for monitoring requests and understanding application behavior. The following example logs each request with timing information:

import time
from ravyn import Ravyn, get
from lilya.middleware import DefineMiddleware

class RequestLoggingMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            start = time.time()
            print(f"[REQ] {scope['method']} {scope['path']}")
            await self.app(scope, receive, send)
            print(f"[RES] {time.time() - start:.3f}s")
        else:
            await self.app(scope, receive, send)

@get("/api/users")
def list_users() -> list:
    return []

app = Ravyn(
    routes=[list_users],
    middleware=[DefineMiddleware(RequestLoggingMiddleware)]
)

This middleware:

  1. Checks if the scope type is HTTP (not WebSocket)
  2. Records the start time before the request is processed
  3. Logs the request method and path
  4. Passes control to the next middleware/handler
  5. Logs the response with total elapsed time in seconds

CORS Configuration

Cross-Origin Resource Sharing (CORS) allows your API to be accessed from different domains. Ravyn provides a CORSConfig object for easy configuration:

from ravyn import Ravyn
from ravyn.config.cors import CORSConfig

app = Ravyn(
    routes=[...],
    cors_config=CORSConfig(
        allow_origins=["https://example.com"],
        allow_methods=["GET", "POST"],
        allow_headers=["Content-Type"]
    )
)

The CORSConfig parameters:

  • allow_origins - List of allowed origin URLs
  • allow_methods - HTTP methods allowed (GET, POST, etc.)
  • allow_headers - Headers the client can send
  • For production, use specific origin URLs instead of "*"

Rate Limiting Pattern

Rate limiting protects your API from abuse by limiting requests per time window. Here's a simple in-memory pattern:

import time
from collections import defaultdict
from ravyn import Ravyn, get
from lilya.middleware import DefineMiddleware
from lilya.responses import PlainText

class SimpleRateLimitMiddleware:
    def __init__(self, app, max_requests=10, window=60):
        self.app = app
        self.max_requests = max_requests
        self.window = window
        self.requests = defaultdict(list)

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        client = scope["client"][0] if scope["client"] else "unknown"
        now = time.time()

        # Clean old requests outside the time window
        self.requests[client] = [
            t for t in self.requests[client]
            if now - t < self.window
        ]

        # Check if limit exceeded
        if len(self.requests[client]) >= self.max_requests:
            response = PlainText("Rate limit exceeded", status_code=429)
            await response(scope, receive, send)
            return

        # Record this request
        self.requests[client].append(now)
        await self.app(scope, receive, send)

app = Ravyn(
    routes=[...],
    middleware=[DefineMiddleware(SimpleRateLimitMiddleware, max_requests=100, window=60)]
)

This middleware:

  1. Tracks requests by client IP address
  2. Removes timestamps older than the time window
  3. Rejects requests if the limit is exceeded (HTTP 429)
  4. Records successful requests in the tracking list

Note

This is a simple in-memory pattern suitable for development and single-server deployments. For production with multiple servers, use Redis or similar solutions.

Middleware Execution Order

Understanding middleware execution order is critical for correct behavior. Middleware executes in the order it's defined, wrapping around the handler:

from ravyn import Ravyn, get
from lilya.middleware import DefineMiddleware

class OuterMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        print("1. Outer: before handler")
        await self.app(scope, receive, send)
        print("4. Outer: after handler")

class InnerMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        print("2. Inner: before handler")
        await self.app(scope, receive, send)
        print("3. Inner: after handler")

@get("/test")
def test() -> dict:
    print("Handler executed")
    return {}

app = Ravyn(
    routes=[test],
    middleware=[
        DefineMiddleware(OuterMiddleware),
        DefineMiddleware(InnerMiddleware),
    ]
)

When a request to /test is made, the execution order is:

1. Outer: before handler
2. Inner: before handler
Handler executed
3. Inner: after handler
4. Outer: after handler

This demonstrates the "onion" pattern where middleware wraps around inner middleware and the handler. The first defined middleware is the outermost layer, and the last defined middleware is closest to the handler.

What is Middleware?

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.

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

Deprecation Notice

BaseAuthMiddleware is deprecated and will be removed in future versions of Ravyn (0.4.0). It is recommended to implement custom authentication by using the the AuthenticationMiddleware instead.

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

RAVYN_SETTINGS_MODULE=configs.live.AppSettings palfrey src:app

INFO:     Palfrey 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"; palfrey src:app

INFO:     Palfrey 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.

AuthenticationMiddleware

This is the new and recommended way of implementing authentication middlewares for Ravyn applications.

It is directly used from Lilya and it is a pure ASGI middleware.

There is a special section for authentication that explains how to use it in more detail.

API Reference

Check out the API Reference for BasseAuthMiddleware for more details.

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.

Troubleshooting

Working with ASGI middleware can sometimes be tricky due to its low-level nature. Here are common issues and how to solve them.

Middleware Order Problems

Problem: Middleware execution order is wrong, causing features like authentication or logging to fail.

Ravyn processes middleware in an "onion" fashion. The first middleware defined is the outermost layer (runs first on request, last on response).

from ravyn import Ravyn, Middleware
from ravyn.middleware.authentication import BasicAuthMiddleware
from myapp.middleware import CustomLoggingMiddleware

# WRONG: Logging won't have user info because auth runs AFTER logging
app = Ravyn(
    middleware=[
        Middleware(CustomLoggingMiddleware),
        Middleware(BasicAuthMiddleware),
    ]
)

# CORRECT: Auth runs first, so Logging can access user data
app = Ravyn(
    middleware=[
        Middleware(BasicAuthMiddleware),
        Middleware(CustomLoggingMiddleware),
    ]
)

Response Already Sent

Problem: Attempting to modify the response or send a new one after the headers have already been sent.

In ASGI, once you send http.response.start, you cannot send it again. If you wrap the send callable, ensure you only call it once for the start message.

async def __call__(self, scope, receive, send):
    async def wrapped_send(message):
        if message["type"] == "http.response.start":
            # Correct: modify headers before sending
            message["headers"].append((b"x-custom", b"value"))
        await send(message)

    await self.app(scope, receive, wrapped_send)

Infinite Loops

Problem: Middleware calling its own path or causing circular dependencies.

This usually happens when a middleware performs a sub-request to the same application using a client that triggers the same middleware.

Solution: Exclude specific paths or check for a custom header to break the loop.

async def __call__(self, scope, receive, send):
    if scope["type"] != "http" or scope["path"] == "/health":
        return await self.app(scope, receive, send)

    # Process other requests
    await self.app(scope, receive, send)

Exception Handling

Problem: Exceptions occurring inside middleware are not caught by the application's standard exception handlers.

Middleware wraps the entire application, including exception handlers. If your middleware raises an exception, it must handle it or it will result in a 500 error that skips your custom handlers.

from ravyn.responses import JSONResponse

async def __call__(self, scope, receive, send):
    try:
        await self.app(scope, receive, send)
    except Exception as exc:
        response = JSONResponse({"error": "Middleware error"}, status_code=500)
        await response(scope, receive, send)

Performance Issues

Problem: Synchronous blocking calls inside async __call__ blocking the entire event loop.

Never use blocking I/O (like requests or standard open()) inside middleware. Always use async equivalents.

# WRONG
import requests

async def __call__(self, scope, receive, send):
    # This blocks the entire server!
    requests.get("https://api.example.com")
    await self.app(scope, receive, send)

# CORRECT
import httpx

async def __call__(self, scope, receive, send):
    async with httpx.AsyncClient() as client:
        await client.get("https://api.example.com")
    await self.app(scope, receive, send)

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.