Skip to content

Request class

Importing the Request from esmerald is as simple as:

from esmerald import Request

esmerald.Request

Request(scope, receive=empty_receive, send=empty_send)

Bases: Request

PARAMETER DESCRIPTION
scope

TYPE: Scope

receive

TYPE: Receive DEFAULT: empty_receive

send

TYPE: Send DEFAULT: empty_send

Source code in esmerald/requests.py
23
24
25
26
27
28
29
30
def __init__(
    self,
    scope: "Scope",
    receive: "Receive" = empty_receive,
    send: "Send" = empty_send,
):
    super().__init__(scope, receive, send)
    self._json: Any = Void

scope instance-attribute

scope = scope

url property

url

base_url property

base_url

headers property

headers

state property

state

query_params property

query_params

path_params property

path_params

cookies property

cookies

client property

client

server property

server

auth property

auth

user property

user

session property

session

is_server_push property

is_server_push

is_server_pull property

is_server_pull

receive property

receive

Get the receive channel of the request.

RETURNS DESCRIPTION
Receive

The receive channel.

TYPE: Receive

send property

send

Get the send of the request.

RETURNS DESCRIPTION
Receive

The send.

TYPE: Send

media property

media

Gathers the information about the media for the request and returns a dictionary type.

charset property

charset

Get a charset for the scope.

content_type property

content_type

Get the content type of the request.

RETURNS DESCRIPTION
str

Tuple[str, Dict[str, str]]: The content type as a tuple containing a string

dict[str, str]

and a dictionary of parameters.

app property

app

method property

method

global_settings property

global_settings

Access to the global settings via request.global_settings.

app_settings property

app_settings

Access to the app settings via request.app_settings.

set_session

set_session(value)

Sets the value of a request session by passing a dictionary.

PARAMETER DESCRIPTION
value

TYPE: Any

Source code in lilya/_internal/_connection.py
175
176
177
178
179
def set_session(self, value: Any) -> None:
    """
    Sets the value of a request session by passing a dictionary.
    """
    self.scope["session"] = value

clear_session

clear_session()

Clears the scope session.

Source code in lilya/_internal/_connection.py
181
182
183
184
185
def clear_session(self) -> None:
    """
    Clears the scope session.
    """
    self.scope["session"] = None

stream async

stream()

Stream the request body in asynchronous chunks.

YIELDS DESCRIPTION
AsyncGenerator[bytes, None]

AsyncGenerator[bytes, None]: Bytes representing chunks of the request body.

Source code in lilya/requests.py
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
async def stream(self) -> AsyncGenerator[bytes, None]:
    """
    Stream the request body in asynchronous chunks.

    Yields:
        AsyncGenerator[bytes, None]: Bytes representing chunks of the request body.
    """
    if self._body is Empty:
        if self._stream_consumed:
            raise RuntimeError("Stream consumed")
        while not self._stream_consumed:
            event = await self._receive()
            if event["type"] == Event.HTTP_REQUEST:
                if not event.get("more_body", False):
                    self._stream_consumed = True
                if event["body"]:
                    yield event["body"]

            elif event["type"] == Event.HTTP_DISCONNECT:
                self._is_disconnected = True
                raise ClientDisconnect()
        yield b""

    else:
        yield cast(bytes, self._body)
        yield b""
        return

body async

body()

Read the entire request body.

RETURNS DESCRIPTION
bytes

The request body as bytes.

TYPE: bytes

Source code in lilya/requests.py
173
174
175
176
177
178
179
180
181
182
async def body(self) -> bytes:
    """
    Read the entire request body.

    Returns:
        bytes: The request body as bytes.
    """
    if self._body is Empty:
        self._body = b"".join([chunk async for chunk in self.stream()])  # type: ignore
    return cast(bytes, self._body)

text async

text()

Returns the body in as a string.

Source code in lilya/requests.py
199
200
201
202
203
204
205
206
207
async def text(self) -> Any:
    """
    Returns the body in as a string.
    """
    body = await self.body()
    try:
        return body.decode(self.charset)
    except (LookupError, ValueError) as exc:
        raise exc

data async

data(*, raise_exception=False)

Returns any form or multipart forms from the request or simply returns a JSON or text/plain format.

PARAMETER DESCRIPTION
raise_exception

TYPE: bool DEFAULT: False

Source code in lilya/requests.py
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
async def data(self, *, raise_exception: bool = False) -> str | bytes | Any:
    """
    Returns any form or multipart forms from the request
    or simply returns a JSON or text/plain format.
    """
    try:
        if self.content_type in (MediaType.MULTIPART, MediaType.URLENCODED):
            return await self.form()
        if self.content_type == MediaType.JSON:
            return await self.json()  # type: ignore
    except ValueError as e:
        if raise_exception:
            raise e
        return await self.body()
    else:
        return await self.text()

form

form(*, max_files=1000, max_fields=1000)

Get the form data from the request.

PARAMETER DESCRIPTION
max_files

TYPE: int | float DEFAULT: 1000

max_fields

TYPE: int | float DEFAULT: 1000

PARAMETER DESCRIPTION
max_files

Maximum number of files allowed in the form data.

TYPE: Union[int, float] DEFAULT: 1000

max_fields

Maximum number of fields allowed in the form data.

TYPE: Union[int, float] DEFAULT: 1000

RETURNS DESCRIPTION
AsyncResourceHandler[FormData]

AsyncResourceHandler[FormData]: Awaiting or using this object will

AsyncResourceHandler[FormData]

return the parsed form data.

Source code in lilya/requests.py
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def form(
    self,
    *,
    max_files: int | float = 1000,
    max_fields: int | float = 1000,
) -> AsyncResourceHandler[FormData]:
    """
    Get the form data from the request.

    Args:
        max_files (Union[int, float]): Maximum number of files allowed
            in the form data.
        max_fields (Union[int, float]): Maximum number of fields allowed
            in the form data.

    Returns:
        AsyncResourceHandler[FormData]: Awaiting or using this object will
        return the parsed form data.
    """
    return AsyncResourceHandler(self._get_form(max_files=max_files, max_fields=max_fields))

close async

close()

Close the request and associated resources.

This includes closing the form data, if any.

Source code in lilya/requests.py
290
291
292
293
294
295
296
297
async def close(self) -> None:
    """
    Close the request and associated resources.

    This includes closing the form data, if any.
    """
    if self._form is not None:
        await self._form.close()

is_disconnected async

is_disconnected()

Check if the client is disconnected.

RETURNS DESCRIPTION
bool

True if the client is disconnected, False otherwise.

TYPE: bool

Source code in lilya/requests.py
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
async def is_disconnected(self) -> bool:
    """
    Check if the client is disconnected.

    Returns:
        bool: True if the client is disconnected, False otherwise.
    """
    if not self._is_disconnected:
        message: Message = {}

        # If message isn't immediately available, move on
        with anyio.CancelScope() as cs:
            cs.cancel()
            message = await self._receive()

        if message.get("type") == "http.disconnect":
            self._is_disconnected = True

    return self._is_disconnected

send_push_promise async

send_push_promise(path)

Send a push promise for the specified path.

PARAMETER DESCRIPTION
path

TYPE: str

PARAMETER DESCRIPTION
path

The path for which to send the push promise.

TYPE: str

Source code in lilya/requests.py
319
320
321
322
323
324
325
326
327
328
329
330
331
async def send_push_promise(self, path: str) -> None:
    """
    Send a push promise for the specified path.

    Args:
        path (str): The path for which to send the push promise.
    """
    if "http.response.push" in self.scope.get("extensions", {}):
        raw_headers: list[tuple[bytes, bytes]] = []
        for name in SERVER_PUSH_HEADERS:
            for value in self.headers.getlist(name):
                raw_headers.append((name.encode("latin-1"), value.encode("latin-1")))
        await self._send({"type": "http.response.push", "path": path, "headers": raw_headers})

json async

json()
Source code in esmerald/requests.py
60
61
62
63
64
65
66
67
async def json(self) -> Any:
    if self._json is Void:
        if "_body" in self.scope:
            body = self.scope["_body"]
        else:
            body = self.scope["_body"] = await self.body() or b"null"
        self._json = loads(body)
    return self._json

path_for

path_for(__name, **path_params)
PARAMETER DESCRIPTION
__name

TYPE: str

**path_params

TYPE: Any DEFAULT: {}

Source code in esmerald/requests.py
69
70
71
def path_for(self, __name: str, **path_params: Any) -> Any:
    url: URL = super().path_for(__name, **path_params)
    return str(url)