Request class

Warning

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

But you can help translating it: Contributing.

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

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

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

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

def clear_session(self) -> None:
    """
    Clears the scope session.
    """
    self.scope["session"] = None

is_secure

is_secure()

Check if the connection is secure (HTTPS).

RETURNS	DESCRIPTION
bool	True if the connection is secure (HTTPS), False otherwise. TYPE: bool

Source code in lilya/_internal/_connection.py

def is_secure(self) -> bool:
    """
    Check if the connection is secure (HTTPS).

    Returns:
        bool: True if the connection is secure (HTTPS), False otherwise.
    """
    return self.url.is_secure

receive async

receive()

The receive channel of the request.

RETURNS	DESCRIPTION
Message	the message. TYPE: Message

Source code in lilya/requests.py

async def receive(self) -> Message:
    """
    The receive channel of the request.

    Returns:
        Message: the message.
    """
    if self._stored_receive_message is not None:
        msg = self._stored_receive_message
        self._stored_receive_message = None
        return msg
    return await self._receive()

sniff async

sniff()

The receive channel of the request.

RETURNS	DESCRIPTION
Message	the message. TYPE: tuple[Message, bool]

Source code in lilya/requests.py

async def sniff(self) -> tuple[Message, bool]:
    """
    The receive channel of the request.

    Returns:
        Message: the message.
    """
    event = await self.receive()
    self._stored_receive_message = event
    more_body = event.get("more_body", False)
    body_is_initialized = False
    if not more_body and event["type"] == Event.HTTP_REQUEST and event["body"]:
        self._body = self.scope["_body"] = event["body"]
        self._stream_consumed = True
        body_is_initialized = True

    return event, body_is_initialized

send async

send(message)

The send of the request.

PARAMETER	DESCRIPTION
message	TYPE: Message

Source code in lilya/requests.py

async def send(self, message: Message) -> None:
    """
    The send of the request.

    """
    await self._send(message)

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

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

async def body(self) -> bytes:
    """
    Read the entire request body.

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

text async

text()

Returns the body in as a string.

Source code in lilya/requests.py

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

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

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

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

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

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("utf-8"), value.encode("utf-8")))
        await self.send({"type": "http.response.push", "path": path, "headers": raw_headers})

json async

json()

Source code in esmerald/requests.py

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

def path_for(self, __name: str, **path_params: Any) -> Any:
    url: URL = super().path_for(__name, **path_params)
    return str(url)