Skip to content

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
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
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
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
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
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
@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
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
@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
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
@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
 8
 9
10
11
12
13
14
15
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
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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 = 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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
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(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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
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(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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
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
11
12
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
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
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
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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
22
23
24
25
26
27
28
29
30
31
32
33
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, send_queue_size=10)

Convert WSGIApp to ASGIApp.

PARAMETER DESCRIPTION
app

TYPE: WSGIApp

workers

TYPE: int DEFAULT: 10

send_queue_size

TYPE: int DEFAULT: 10

Source code in .venv/lib/python3.8/site-packages/a2wsgi/wsgi.py
153
154
155
156
157
158
159
160
def __init__(
    self, app: WSGIApp, workers: int = 10, send_queue_size: int = 10
) -> None:
    self.app = app
    self.send_queue_size = send_queue_size
    self.executor = ThreadPoolExecutor(
        thread_name_prefix="WSGI", max_workers=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)