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

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

query_params property

query_params

path_params property

path_params

cookies property

cookies

client property

client

session property

session

auth property

auth

user property

user

state property

state

receive property

receive

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.

stream async

stream()

Source code in .venv/lib/python3.8/site-packages/starlette/requests.py

async def stream(self) -> typing.AsyncGenerator[bytes, None]:
    if hasattr(self, "_body"):
        yield self._body
        yield b""
        return
    if self._stream_consumed:
        raise RuntimeError("Stream consumed")
    while not self._stream_consumed:
        message = await self._receive()
        if message["type"] == "http.request":
            body = message.get("body", b"")
            if not message.get("more_body", False):
                self._stream_consumed = True
            if body:
                yield body
        elif message["type"] == "http.disconnect":
            self._is_disconnected = True
            raise ClientDisconnect()
    yield b""

body async

body()

Source code in .venv/lib/python3.8/site-packages/starlette/requests.py

async def body(self) -> bytes:
    if not hasattr(self, "_body"):
        chunks: "typing.List[bytes]" = []
        async for chunk in self.stream():
            chunks.append(chunk)
        self._body = b"".join(chunks)
    return self._body

form

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

PARAMETER	DESCRIPTION
max_files	TYPE: Union[int, float] DEFAULT: 1000
max_fields	TYPE: Union[int, float] DEFAULT: 1000

Source code in .venv/lib/python3.8/site-packages/starlette/requests.py

def form(
    self,
    *,
    max_files: typing.Union[int, float] = 1000,
    max_fields: typing.Union[int, float] = 1000,
) -> AwaitableOrContextManager[FormData]:
    return AwaitableOrContextManagerWrapper(
        self._get_form(max_files=max_files, max_fields=max_fields)
    )

close async

close()

Source code in .venv/lib/python3.8/site-packages/starlette/requests.py

async def close(self) -> None:
    if self._form is not None:
        await self._form.close()

is_disconnected async

is_disconnected()

Source code in .venv/lib/python3.8/site-packages/starlette/requests.py

async def is_disconnected(self) -> bool:
    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)

PARAMETER	DESCRIPTION
path	TYPE: str

Source code in .venv/lib/python3.8/site-packages/starlette/requests.py

async def send_push_promise(self, path: str) -> None:
    if "http.response.push" in self.scope.get("extensions", {}):
        raw_headers: "typing.List[typing.Tuple[bytes, bytes]]" = []
        for name in SERVER_PUSH_HEADERS_TO_COPY:
            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

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

url_for

url_for(__name, **path_params)

PARAMETER	DESCRIPTION
__name	TYPE: str
**path_params	TYPE: Any DEFAULT: {}

Source code in esmerald/requests.py

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