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, 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
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 .venv/lib/python3.8/site-packages/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,
    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_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,
    )

    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

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

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 .venv/lib/python3.8/site-packages/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 .venv/lib/python3.8/site-packages/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)

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 .venv/lib/python3.8/site-packages/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 .venv/lib/python3.8/site-packages/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 .venv/lib/python3.8/site-packages/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)

Get headers for simple (non-preflight) responses.

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

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 .venv/lib/python3.8/site-packages/lilya/middleware/cors.py

@staticmethod
def get_simple_headers(
    allow_all_origins: bool, allow_credentials: bool, expose_headers: Sequence[str]
) -> 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)

    return simple_headers

get_preflight_headers staticmethod

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

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

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 .venv/lib/python3.8/site-packages/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,
) -> 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_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 .venv/lib/python3.8/site-packages/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 .venv/lib/python3.8/site-packages/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 .venv/lib/python3.8/site-packages/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 .venv/lib/python3.8/site-packages/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 .venv/lib/python3.8/site-packages/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 .venv/lib/python3.8/site-packages/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 .venv/lib/python3.8/site-packages/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 .venv/lib/python3.8/site-packages/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 .venv/lib/python3.8/site-packages/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 .venv/lib/python3.8/site-packages/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 .venv/lib/python3.8/site-packages/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)