Middlewares

All the Esmerald native middlewares.

esmerald.middleware.CORSMiddleware

CORSMiddleware(app, allow_origins=None, allow_methods=None, allow_headers=None, allow_credentials=False, allow_origin_regex=None, allow_private_networks=False, expose_headers=None, max_age=600)

Bases: MiddlewareProtocol

Middleware for handling Cross-Origin Resource Sharing (CORS) headers.

PARAMETER	DESCRIPTION
app	TYPE: ASGIApp
allow_origins	TYPE: Sequence[str] | None DEFAULT: None
allow_methods	TYPE: Sequence[str] | None DEFAULT: None
allow_headers	TYPE: Sequence[str] | None DEFAULT: None
allow_credentials	TYPE: bool DEFAULT: False
allow_origin_regex	TYPE: str | None DEFAULT: None
allow_private_networks	TYPE: bool DEFAULT: False
expose_headers	TYPE: Sequence[str] | None DEFAULT: None
max_age	TYPE: int DEFAULT: 600

PARAMETER	DESCRIPTION
app	The ASGI application to wrap. TYPE: ASGIApp
allow_origins	List of allowed origin patterns. TYPE: Sequence[str] DEFAULT: None
allow_methods	List of allowed HTTP methods. TYPE: Sequence[str] DEFAULT: None
allow_headers	List of allowed HTTP headers. TYPE: Sequence[str] DEFAULT: None
allow_credentials	Whether credentials such as cookies are allowed. TYPE: bool DEFAULT: False
allow_origin_regex	Regular expression for allowed origins. TYPE: Optional[str] DEFAULT: None
expose_headers	List of headers exposed to the browser. TYPE: Sequence[str] DEFAULT: None
max_age	Maximum age (in seconds) for caching preflight requests. TYPE: int DEFAULT: 600

Source code in lilya/middleware/cors.py

def __init__(
    self,
    app: ASGIApp,
    allow_origins: Sequence[str] | None = None,
    allow_methods: Sequence[str] | None = None,
    allow_headers: Sequence[str] | None = None,
    allow_credentials: bool = False,
    allow_origin_regex: str | None = None,
    allow_private_networks: bool = False,
    expose_headers: Sequence[str] | None = None,
    max_age: int = 600,
) -> None:
    """
    Middleware for handling Cross-Origin Resource Sharing (CORS) headers.

    Args:
        app (ASGIApp): The ASGI application to wrap.
        allow_origins (Sequence[str]): List of allowed origin patterns.
        allow_methods (Sequence[str]): List of allowed HTTP methods.
        allow_headers (Sequence[str]): List of allowed HTTP headers.
        allow_credentials (bool): Whether credentials such as cookies are allowed.
        allow_origin_regex (Optional[str]): Regular expression for allowed origins.
        expose_headers (Sequence[str]): List of headers exposed to the browser.
        max_age (int): Maximum age (in seconds) for caching preflight requests.
    """
    allow_origins = allow_origins or ()
    allow_methods = allow_methods or ("GET",)
    allow_headers = allow_headers or ()
    expose_headers = expose_headers or ()

    if "*" in allow_methods:
        allow_methods = HTTPCorsEnum.to_tuple()

    compiled_allow_origin_regex = None
    if allow_origin_regex is not None:
        compiled_allow_origin_regex = re.compile(allow_origin_regex)

    allow_all_origins = "*" in allow_origins
    allow_all_headers = "*" in allow_headers
    preflight_explicit_allow_origin = not allow_all_origins or allow_credentials

    simple_headers = self.get_simple_headers(
        allow_all_origins, allow_credentials, expose_headers, allow_private_networks
    )

    allow_headers = sorted(HeaderEnum.to_set() | set(allow_headers))
    preflight_headers = self.get_preflight_headers(
        allow_all_origins,
        allow_methods,
        max_age,
        allow_all_headers,
        allow_headers,
        allow_credentials,
        allow_private_networks,
    )

    self.app = app
    self.allow_origins = allow_origins
    self.allow_methods = allow_methods
    self.allow_headers = [h.lower() for h in allow_headers]
    self.allow_all_origins = allow_all_origins
    self.allow_all_headers = allow_all_headers
    self.preflight_explicit_allow_origin = preflight_explicit_allow_origin
    self.allow_origin_regex = compiled_allow_origin_regex
    self.simple_headers = simple_headers
    self.preflight_headers = preflight_headers
    self.allow_private_networks = allow_private_networks

app instance-attribute

app = app

allow_origins instance-attribute

allow_origins = allow_origins

allow_methods instance-attribute

allow_methods = allow_methods

allow_headers instance-attribute

allow_headers = [lower() for h in allow_headers]

allow_all_origins instance-attribute

allow_all_origins = allow_all_origins

allow_all_headers instance-attribute

allow_all_headers = allow_all_headers

preflight_explicit_allow_origin instance-attribute

preflight_explicit_allow_origin = preflight_explicit_allow_origin

allow_origin_regex instance-attribute

allow_origin_regex = compiled_allow_origin_regex

simple_headers instance-attribute

simple_headers = simple_headers

preflight_headers instance-attribute

preflight_headers = preflight_headers

allow_private_networks instance-attribute

allow_private_networks = allow_private_networks

validate_origin

validate_origin(origin)

Validate if the origin is allowed.

PARAMETER	DESCRIPTION
origin	TYPE: str

PARAMETER	DESCRIPTION
origin	Origin header value. TYPE: str

RETURNS	DESCRIPTION
bool	True if the origin is allowed, False otherwise. TYPE: bool

Source code in lilya/middleware/cors.py

def validate_origin(self, origin: str) -> bool:
    """
    Validate if the origin is allowed.

    Args:
        origin (str): Origin header value.

    Returns:
        bool: True if the origin is allowed, False otherwise.
    """
    if self.allow_all_origins:
        return True

    if self.allow_origin_regex is not None and self.allow_origin_regex.fullmatch(origin):
        return True

    return origin in self.allow_origins

preflight_response

preflight_response(request_headers)

Generate a preflight response.

PARAMETER	DESCRIPTION
request_headers	TYPE: Header

PARAMETER	DESCRIPTION
request_headers	Header from the incoming preflight request. TYPE: Header

RETURNS	DESCRIPTION
Response	Preflight response. TYPE: Response

Source code in lilya/middleware/cors.py

def preflight_response(self, request_headers: Header) -> Response:
    """
    Generate a preflight response.

    Args:
        request_headers (Header): Header from the incoming preflight request.

    Returns:
        Response: Preflight response.
    """
    requested_origin = request_headers["origin"]
    requested_method = request_headers["access-control-request-method"]
    requested_headers = request_headers.get("access-control-request-headers")

    headers = dict(self.preflight_headers)
    failures = []

    if self.validate_origin(origin=requested_origin):
        if self.preflight_explicit_allow_origin:
            headers["Access-Control-Allow-Origin"] = requested_origin
    else:
        failures.append("origin")

    if requested_method not in self.allow_methods:
        failures.append("method")

    if self.allow_all_headers and requested_headers is not None:
        headers["Access-Control-Allow-Headers"] = requested_headers
    elif requested_headers is not None:
        for header in [h.lower() for h in requested_headers.split(",")]:
            if header.strip() not in self.allow_headers:
                failures.append("headers")
                break

    if failures:
        failure_text = "Disallowed CORS " + ", ".join(failures)
        return PlainText(failure_text, status_code=400, headers=headers)

    return PlainText("OK", status_code=200, headers=headers)

preflight_private_network_response

preflight_private_network_response(request_headers)

Process the preflight request for private network access. This feature is not part of the CORS specification but is useful for the new browsers that enforce the private network access policy.

By the time of this implementation, this specific functionality got inspired by great developers thinking about the future of the web and the security of the users.

More information about the private network access policy can be found at:

1. https://developer.chrome.com/blog/private-network-access-preflight/
2. https://documentation.alphasoftware.com/documentation/pages/Guides/Mobile%20and%20Web%20Components/UX/Properties/Advanced/CORS%20allow%20private%20network.xml#:~:text=Enables%20cross%20origin%20requests%20from,(e.g.%20behind%20a%20firewall).

This will be rolled out on Chromium based browsers such as Google Chrome, Microsoft Edge, Brave and others.

PARAMETER	DESCRIPTION
request_headers	TYPE: Header

PARAMETER	DESCRIPTION
request_headers	The headers of the request. TYPE: Header

RETURNS	DESCRIPTION
Response	The response to the preflight request. TYPE: Response

Source code in lilya/middleware/cors.py

def preflight_private_network_response(self, request_headers: Header) -> Response:
    """
    Process the preflight request for private network access.
    This feature is not part of the CORS specification but is useful for the new browsers
    that enforce the private network access policy.

    By the time of this implementation, this specific functionality got inspired by great
    developers thinking about the future of the web and the security of the users.

    More information about the private network access policy can be found at:

        1. https://developer.chrome.com/blog/private-network-access-preflight/
        2. https://documentation.alphasoftware.com/documentation/pages/Guides/Mobile%20and%20Web%20Components/UX/Properties/Advanced/CORS%20allow%20private%20network.xml#:~:text=Enables%20cross%20origin%20requests%20from,(e.g.%20behind%20a%20firewall).

    This will be rolled out on Chromium based browsers such as Google Chrome, Microsoft Edge, Brave and others.

    Args:
        request_headers (Header): The headers of the request.

    Returns:
        Response: The response to the preflight request.
    """
    requested_origin = request_headers["origin"]
    requested_private_network = request_headers["access-control-request-private-network"]

    headers = dict(self.preflight_headers)
    errors = []

    if not self.validate_origin(origin=requested_origin):
        errors.append("origin")

    if requested_private_network == "true" and not self.allow_private_networks:
        errors.append("private-network")

    if errors:
        message = "Disallowed Private Network Access " + ", ".join(errors)
        return PlainText(message, status_code=400, headers=headers)

    headers["Access-Control-Allow-Origin"] = requested_origin
    return PlainText("Allowed", status_code=200, headers=headers)

simple_response async

simple_response(scope, receive, send, request_headers)

Handle the simple response.

PARAMETER	DESCRIPTION
scope	TYPE: Scope
receive	TYPE: Receive
send	TYPE: Send
request_headers	TYPE: Header

PARAMETER	DESCRIPTION
scope	ASGI scope. TYPE: Scope
receive	ASGI receive channel. TYPE: Receive
send	ASGI send channel. TYPE: Send
request_headers	Header from the incoming request. TYPE: Header

Source code in lilya/middleware/cors.py

async def simple_response(
    self, scope: Scope, receive: Receive, send: Send, request_headers: Header
) -> None:
    """
    Handle the simple response.

    Args:
        scope (Scope): ASGI scope.
        receive (Receive): ASGI receive channel.
        send (Send): ASGI send channel.
        request_headers (Header): Header from the incoming request.
    """
    send = functools.partial(self.send, send=send, request_headers=request_headers)
    await self.app(scope, receive, send)

send async

send(message, send, request_headers)

Send the message and apply CORS headers.

PARAMETER	DESCRIPTION
message	TYPE: Message
send	TYPE: Send
request_headers	TYPE: Header

PARAMETER	DESCRIPTION
message	ASGI message. TYPE: Message
send	ASGI send channel. TYPE: Send
request_headers	Header from the incoming request. TYPE: Header

Source code in lilya/middleware/cors.py

async def send(self, message: Message, send: Send, request_headers: Header) -> None:
    """
    Send the message and apply CORS headers.

    Args:
        message (Message): ASGI message.
        send (Send): ASGI send channel.
        request_headers (Header): Header from the incoming request.
    """
    if message["type"] != "http.response.start":
        await send(message)
        return

    message.setdefault("headers", [])
    headers = Header.from_scope(scope=message)
    headers.update(self.simple_headers)
    origin = request_headers["Origin"]
    has_cookie = "cookie" in request_headers

    if self.allow_all_origins and has_cookie:
        self.set_explicit_origin(headers, origin)
    elif not self.allow_all_origins and self.validate_origin(origin=origin):
        self.set_explicit_origin(headers, origin)

    message["headers"] = headers.get_multi_items()
    await send(message)

set_explicit_origin staticmethod

set_explicit_origin(headers, origin)

Set explicit origin in headers.

PARAMETER	DESCRIPTION
headers	TYPE: Header
origin	TYPE: str

PARAMETER	DESCRIPTION
headers	MutableHeaders instance. TYPE: MutableHeaders
origin	Origin header value. TYPE: str

Source code in lilya/middleware/cors.py

@staticmethod
def set_explicit_origin(headers: Header, origin: str) -> None:
    """
    Set explicit origin in headers.

    Args:
        headers (MutableHeaders): MutableHeaders instance.
        origin (str): Origin header value.
    """
    headers["Access-Control-Allow-Origin"] = origin
    headers.add_vary_header("Origin")

    expose_headers = headers.get("Access-Control-Expose-Headers")
    if expose_headers:
        headers.add_vary_header("Access-Control-Expose-Headers")

get_simple_headers staticmethod

get_simple_headers(allow_all_origins, allow_credentials, expose_headers, allow_private_networks)

Get headers for simple (non-preflight) responses.

PARAMETER	DESCRIPTION
allow_all_origins	TYPE: bool
allow_credentials	TYPE: bool
expose_headers	TYPE: Sequence[str]
allow_private_networks	TYPE: bool

PARAMETER	DESCRIPTION
allow_all_origins	Whether all origins are allowed. TYPE: bool
allow_credentials	Whether credentials such as cookies are allowed. TYPE: bool
expose_headers	List of headers exposed to the browser. TYPE: Sequence[str]

RETURNS	DESCRIPTION
dict	Dictionary of simple response headers. TYPE: dict

Source code in lilya/middleware/cors.py

@staticmethod
def get_simple_headers(
    allow_all_origins: bool,
    allow_credentials: bool,
    expose_headers: Sequence[str],
    allow_private_networks: bool,
) -> dict:
    """
    Get headers for simple (non-preflight) responses.

    Args:
        allow_all_origins (bool): Whether all origins are allowed.
        allow_credentials (bool): Whether credentials such as cookies are allowed.
        expose_headers (Sequence[str]): List of headers exposed to the browser.

    Returns:
        dict: Dictionary of simple response headers.
    """
    simple_headers = {}

    if allow_all_origins:
        simple_headers["Access-Control-Allow-Origin"] = "*"
    if allow_credentials:
        simple_headers["Access-Control-Allow-Credentials"] = "true"
    if expose_headers:
        simple_headers["Access-Control-Expose-Headers"] = ", ".join(expose_headers)
    if allow_private_networks:
        simple_headers["Access-Control-Allow-Private-Network"] = "true"

    return simple_headers

get_preflight_headers staticmethod

get_preflight_headers(allow_all_origins, allow_methods, max_age, allow_all_headers, allow_headers, allow_credentials, allow_private_networks)

Get headers for preflight responses.

PARAMETER	DESCRIPTION
allow_all_origins	TYPE: bool
allow_methods	TYPE: Sequence[str]
max_age	TYPE: int
allow_all_headers	TYPE: bool
allow_headers	TYPE: Sequence[str]
allow_credentials	TYPE: bool
allow_private_networks	TYPE: bool

PARAMETER	DESCRIPTION
allow_all_origins	Whether all origins are allowed. TYPE: bool
allow_methods	List of allowed HTTP methods. TYPE: Sequence[str]
max_age	Maximum age (in seconds) for caching preflight requests. TYPE: int
allow_all_headers	Whether all headers are allowed. TYPE: bool
allow_headers	List of allowed HTTP headers. TYPE: Sequence[str]
allow_credentials	Whether credentials such as cookies are allowed. TYPE: bool

RETURNS	DESCRIPTION
dict	Dictionary of preflight response headers. TYPE: dict[str, Any]

Source code in lilya/middleware/cors.py

@staticmethod
def get_preflight_headers(
    allow_all_origins: bool,
    allow_methods: Sequence[str],
    max_age: int,
    allow_all_headers: bool,
    allow_headers: Sequence[str],
    allow_credentials: bool,
    allow_private_networks: bool,
) -> dict[str, Any]:
    """
    Get headers for preflight responses.

    Args:
        allow_all_origins (bool): Whether all origins are allowed.
        allow_methods (Sequence[str]): List of allowed HTTP methods.
        max_age (int): Maximum age (in seconds) for caching preflight requests.
        allow_all_headers (bool): Whether all headers are allowed.
        allow_headers (Sequence[str]): List of allowed HTTP headers.
        allow_credentials (bool): Whether credentials such as cookies are allowed.

    Returns:
        dict: Dictionary of preflight response headers.
    """
    preflight_headers = {}
    explicit_allow_origin = not allow_all_origins or allow_credentials

    if allow_all_origins:
        preflight_headers["Access-Control-Allow-Origin"] = "*"
    if allow_credentials:
        preflight_headers["Access-Control-Allow-Credentials"] = "true"
    if allow_private_networks:
        preflight_headers["Access-Control-Allow-Private-Network"] = "true"
    if allow_all_headers:
        preflight_headers["Access-Control-Allow-Headers"] = "*"
    elif allow_headers:
        preflight_headers["Access-Control-Allow-Headers"] = ", ".join(allow_headers)

    if explicit_allow_origin:
        preflight_headers["Vary"] = "Origin"

    preflight_headers.update(
        {
            "Access-Control-Allow-Methods": ", ".join(allow_methods),
            "Access-Control-Max-Age": str(max_age),
        }
    )

    return preflight_headers

esmerald.middleware.RequestSettingsMiddleware

RequestSettingsMiddleware(app)

Bases: MiddlewareProtocol

Settings Middleware class.

PARAMETER	DESCRIPTION
app	TYPE: ASGIApp

PARAMETER	DESCRIPTION
app	The 'next' ASGI app to call. TYPE: ASGIApp

Source code in esmerald/middleware/settings_middleware.py

def __init__(self, app: "ASGIApp"):
    """Settings Middleware class.

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

app instance-attribute

app = app

esmerald.middleware.SessionMiddleware

SessionMiddleware(app, secret_key, session_cookie='session', max_age=14 * 24 * 60 * 60, path='/', same_site='lax', https_only=False, domain=None)

Bases: MiddlewareProtocol

Middleware for handling session data in ASGI applications.

PARAMETER	DESCRIPTION
app	TYPE: ASGIApp
secret_key	TYPE: str | Secret
session_cookie	TYPE: str DEFAULT: 'session'
max_age	TYPE: int | None DEFAULT: 14 * 24 * 60 * 60
path	TYPE: str DEFAULT: '/'
same_site	TYPE: Literal['lax', 'strict', 'none'] DEFAULT: 'lax'
https_only	TYPE: bool DEFAULT: False
domain	TYPE: str | None DEFAULT: None

PARAMETER	DESCRIPTION
app	The ASGI application to wrap. TYPE: ASGIApp
secret_key	The secret key used for signing session data. TYPE: Union[str, Secret]
session_cookie	The name of the session cookie. TYPE: str DEFAULT: 'session'
max_age	The maximum age of the session in seconds (default is 14 days). TYPE: Optional[int] DEFAULT: 14 * 24 * 60 * 60
path	The path attribute for the session cookie. TYPE: str DEFAULT: '/'
same_site	The SameSite attribute for the session cookie. TYPE: Literal['lax', 'strict', 'none'] DEFAULT: 'lax'
https_only	If True, set the secure flag for the session cookie (HTTPS only). TYPE: bool DEFAULT: False
domain	The domain attribute for the session cookie. TYPE: Optional[str] DEFAULT: None

Source code in lilya/middleware/sessions.py

def __init__(
    self,
    app: ASGIApp,
    secret_key: str | Secret,
    session_cookie: str = "session",
    max_age: int | None = 14 * 24 * 60 * 60,  # 14 days, in seconds
    path: str = "/",
    same_site: Literal["lax", "strict", "none"] = "lax",
    https_only: bool = False,
    domain: str | None = None,
) -> None:
    """
    Middleware for handling session data in ASGI applications.

    Args:
        app (ASGIApp): The ASGI application to wrap.
        secret_key (Union[str, Secret]): The secret key used for signing session data.
        session_cookie (str): The name of the session cookie.
        max_age (Optional[int]): The maximum age of the session in seconds (default is 14 days).
        path (str): The path attribute for the session cookie.
        same_site (Literal["lax", "strict", "none"]): The SameSite attribute for the session cookie.
        https_only (bool): If True, set the secure flag for the session cookie (HTTPS only).
        domain (Optional[str]): The domain attribute for the session cookie.
    """
    self.app = app
    self.signer = itsdangerous.TimestampSigner(str(secret_key))
    self.session_cookie = session_cookie
    self.max_age = max_age
    self.path = path
    self.security_flags = "httponly; samesite=" + same_site
    if https_only:
        self.security_flags += "; secure"
    if domain is not None:
        self.security_flags += f"; domain={domain}"

app instance-attribute

app = app

signer instance-attribute

signer = TimestampSigner(str(secret_key))

session_cookie instance-attribute

session_cookie = session_cookie

max_age instance-attribute

max_age = max_age

path instance-attribute

path = path

security_flags instance-attribute

security_flags = 'httponly; samesite=' + same_site

load_session_data async

load_session_data(scope, connection)

Load session data from the session cookie.

PARAMETER	DESCRIPTION
scope	TYPE: Scope
connection	TYPE: Connection

PARAMETER	DESCRIPTION
scope	ASGI scope. TYPE: Scope
connection	HTTP connection object. TYPE: Connection

RETURNS	DESCRIPTION
bool	True if the initial session was empty, False otherwise. TYPE: bool

Source code in lilya/middleware/sessions.py

async def load_session_data(self, scope: Scope, connection: Connection) -> bool:
    """
    Load session data from the session cookie.

    Args:
        scope (Scope): ASGI scope.
        connection (Connection): HTTP connection object.

    Returns:
        bool: True if the initial session was empty, False otherwise.
    """
    if self.session_cookie in connection.cookies:
        data = connection.cookies[self.session_cookie].encode("utf-8")
        try:
            data = self.signer.unsign(data, max_age=self.max_age)
            scope["session"] = json.loads(b64decode(data))
            return False
        except BadSignature:
            scope["session"] = {}
    else:
        scope["session"] = {}
    return True

process_response async

process_response(message, scope, initial_session_was_empty, send)

Process the response and set the session cookie.

PARAMETER	DESCRIPTION
message	TYPE: Message
scope	TYPE: Scope
initial_session_was_empty	TYPE: bool
send	TYPE: Send

PARAMETER	DESCRIPTION
message	ASGI message. TYPE: Message
scope	ASGI scope. TYPE: Scope
initial_session_was_empty	True if the initial session was empty, False otherwise. TYPE: bool
send	ASGI send channel. TYPE: Send

Source code in lilya/middleware/sessions.py

async def process_response(
    self,
    message: Message,
    scope: Scope,
    initial_session_was_empty: bool,
    send: Send,
) -> None:
    """
    Process the response and set the session cookie.

    Args:
        message (Message): ASGI message.
        scope (Scope): ASGI scope.
        initial_session_was_empty (bool): True if the initial session was empty, False otherwise.
        send (Send): ASGI send channel.
    """
    if message["type"] == "http.response.start":
        if scope["session"]:
            message = await self.set_session_cookie(scope, message)
        elif not initial_session_was_empty:
            message = await self.clear_session_cookie(scope, message)

    await send(message)

set_session_cookie async

set_session_cookie(scope, message)

Set the session cookie in the response headers.

PARAMETER	DESCRIPTION
scope	TYPE: Scope
message	TYPE: Message

PARAMETER	DESCRIPTION
scope	ASGI scope. TYPE: Scope
message	ASGI message TYPE: Message

Source code in lilya/middleware/sessions.py

async def set_session_cookie(self, scope: Scope, message: Message) -> Message:
    """
    Set the session cookie in the response headers.

    Args:
        scope (Scope): ASGI scope.
        message (Message): ASGI message
    """
    data = b64encode(json.dumps(scope["session"]).encode("utf-8"))
    data = self.signer.sign(data)
    headers = Header.from_scope(scope=scope)
    header_value = "{session_cookie}={data}; path={path}; {max_age}{security_flags}".format(
        session_cookie=self.session_cookie,
        data=data.decode("utf-8"),
        path=self.path,
        max_age=f"Max-Age={self.max_age}; " if self.max_age else "",
        security_flags=self.security_flags,
    )
    headers.add("Set-Cookie", header_value)
    message["headers"] = headers.get_multi_items()
    return message

clear_session_cookie async

clear_session_cookie(scope, message)

Clear the session cookie in the response headers.

PARAMETER	DESCRIPTION
scope	TYPE: Scope
message	TYPE: Message

PARAMETER	DESCRIPTION
scope	ASGI scope. TYPE: Scope

Source code in lilya/middleware/sessions.py

async def clear_session_cookie(self, scope: Scope, message: Message) -> Message:
    """
    Clear the session cookie in the response headers.

    Args:
        scope (Scope): ASGI scope.
    """
    headers = Header.from_scope(scope=scope)
    header_value = "{session_cookie}={data}; path={path}; {expires}{security_flags}".format(
        session_cookie=self.session_cookie,
        data="null",
        path=self.path,
        expires="expires=Thu, 01 Jan 1970 00:00:00 GMT; ",
        security_flags=self.security_flags,
    )
    headers.add("Set-Cookie", header_value)
    message["headers"] = headers.get_multi_items()
    return message

esmerald.middleware.HTTPSRedirectMiddleware

HTTPSRedirectMiddleware(app)

Bases: MiddlewareProtocol

PARAMETER	DESCRIPTION
app	TYPE: ASGIApp

Source code in lilya/middleware/httpsredirect.py

def __init__(self, app: ASGIApp) -> None:
    self.app = app

app instance-attribute

app = app

esmerald.middleware.TrustedHostMiddleware

TrustedHostMiddleware(app, allowed_hosts=None, www_redirect=True)

Bases: MiddlewareProtocol

Middleware for enforcing trusted host headers in incoming requests.

PARAMETER	DESCRIPTION
app	TYPE: ASGIApp
allowed_hosts	TYPE: Sequence[str] | None DEFAULT: None
www_redirect	TYPE: bool DEFAULT: True

PARAMETER	DESCRIPTION
app	The ASGI application to wrap. TYPE: ASGIApp
allowed_hosts	List of allowed host patterns. Defaults to ["*"]. TYPE: Optional[Sequence[str]] DEFAULT: None
www_redirect	Whether to redirect requests with missing 'www.' to 'www.' prefixed URLs. TYPE: bool DEFAULT: True

Source code in lilya/middleware/trustedhost.py

def __init__(
    self,
    app: ASGIApp,
    allowed_hosts: typing.Sequence[str] | None = None,
    www_redirect: bool = True,
) -> None:
    """
    Middleware for enforcing trusted host headers in incoming requests.

    Args:
        app (ASGIApp): The ASGI application to wrap.
        allowed_hosts (Optional[Sequence[str]]): List of allowed host patterns. Defaults to ["*"].
        www_redirect (bool): Whether to redirect requests with missing 'www.' to 'www.' prefixed URLs.
    """
    if allowed_hosts is None:
        allowed_hosts = ["*"]

    for pattern in allowed_hosts:
        assert "*" not in pattern[1:], ENFORCE_DOMAIN_WILDCARD
        if pattern.startswith("*") and pattern != "*":
            assert pattern.startswith("*."), ENFORCE_DOMAIN_WILDCARD

    self.app = app
    self.allowed_hosts = list(allowed_hosts)
    self.allow_any = "*" in allowed_hosts
    self.www_redirect = www_redirect

app instance-attribute

app = app

allowed_hosts instance-attribute

allowed_hosts = list(allowed_hosts)

allow_any instance-attribute

allow_any = '*' in allowed_hosts

www_redirect instance-attribute

www_redirect = www_redirect

validate_host

validate_host(host)

Validate the host header against the allowed host patterns.

PARAMETER	DESCRIPTION
host	TYPE: str

PARAMETER	DESCRIPTION
host	Host header value. TYPE: str

RETURNS	DESCRIPTION
tuple[bool, bool]	Tuple[bool, bool]: (is_valid_host, found_www_redirect).

Source code in lilya/middleware/trustedhost.py

def validate_host(self, host: str) -> tuple[bool, bool]:
    """
    Validate the host header against the allowed host patterns.

    Args:
        host (str): Host header value.

    Returns:
        Tuple[bool, bool]: (is_valid_host, found_www_redirect).
    """
    is_valid_host = False
    found_www_redirect = False
    for pattern in self.allowed_hosts:
        if host == pattern or (pattern.startswith("*") and host.endswith(pattern[1:])):
            is_valid_host = True
            break
        elif "www." + host == pattern:
            found_www_redirect = True
    return is_valid_host, found_www_redirect

handle_invalid_host async

handle_invalid_host(scope, receive, send, found_www_redirect)

Handle requests with invalid host headers.

PARAMETER	DESCRIPTION
scope	TYPE: Scope
receive	TYPE: Receive
send	TYPE: Send
found_www_redirect	TYPE: bool

PARAMETER	DESCRIPTION
scope	ASGI scope. TYPE: Scope
receive	ASGI receive channel. TYPE: Receive
send	ASGI send channel. TYPE: Send
found_www_redirect	Whether 'www.' redirect was found. TYPE: bool

Source code in lilya/middleware/trustedhost.py

async def handle_invalid_host(
    self, scope: Scope, receive: Receive, send: Send, found_www_redirect: bool
) -> None:
    """
    Handle requests with invalid host headers.

    Args:
        scope (Scope): ASGI scope.
        receive (Receive): ASGI receive channel.
        send (Send): ASGI send channel.
        found_www_redirect (bool): Whether 'www.' redirect was found.
    """
    response: Response
    if found_www_redirect and self.www_redirect:
        url = URL.build_from_scope(scope=scope)
        redirect_url = url.replace(netloc="www." + url.netloc)
        response = RedirectResponse(url=str(redirect_url))
    else:
        response = PlainText("Invalid host header", status_code=400)
    await response(scope, receive, send)

esmerald.middleware.GZipMiddleware

GZipMiddleware(app, minimum_size=500, compresslevel=9)

Bases: MiddlewareProtocol

Middleware to compress responses with GZip.

PARAMETER	DESCRIPTION
app	The 'next' ASGI app to call. TYPE: ASGIApp
minimum_size	Minimum response size to trigger compression. TYPE: int DEFAULT: 500
compresslevel	GZip compression level (0 to 9). TYPE: int DEFAULT: 9

Initialize GZipMiddleware.

PARAMETER	DESCRIPTION
app	TYPE: ASGIApp
minimum_size	TYPE: int DEFAULT: 500
compresslevel	TYPE: int DEFAULT: 9

PARAMETER	DESCRIPTION
app	The 'next' ASGI app to call. TYPE: ASGIApp
minimum_size	Minimum response size to trigger compression. TYPE: int DEFAULT: 500
compresslevel	GZip compression level (0 to 9). TYPE: int DEFAULT: 9

Source code in lilya/middleware/compression.py

def __init__(self, app: ASGIApp, minimum_size: int = 500, compresslevel: int = 9) -> None:
    """
    Initialize GZipMiddleware.

    Args:
        app: The 'next' ASGI app to call.
        minimum_size: Minimum response size to trigger compression.
        compresslevel: GZip compression level (0 to 9).
    """
    self.app = app
    self.minimum_size = minimum_size
    self.compresslevel = compresslevel

app instance-attribute

app = app

minimum_size instance-attribute

minimum_size = minimum_size

compresslevel instance-attribute

compresslevel = compresslevel

esmerald.middleware.wsgi.WSGIMiddleware

WSGIMiddleware(app, workers=10)

Bases: WSGIMiddleware

PARAMETER	DESCRIPTION
app	TYPE: WSGIApp | str
workers	TYPE: int DEFAULT: 10

Source code in lilya/middleware/wsgi.py

def __init__(self, app: WSGIApp | str, workers: int = 10) -> None:
    if isinstance(app, str):
        app = cast(WSGIApp, import_string(app))
    super().__init__(app, workers)

app instance-attribute

app = app

send_queue_size instance-attribute

send_queue_size = send_queue_size

executor instance-attribute

executor = ThreadPoolExecutor(thread_name_prefix='WSGI', max_workers=workers)