Skip to content

AsgiRequest

faststream.asgi.request.AsgiRequest #

AsgiRequest(scope: Scope, receive: Receive, send: Send)
Source code in faststream/asgi/request.py 
16
17
18
19
20
21
22
23
24
def __init__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
    self._scope = scope
    self._receive = receive
    self._send = send
    self._stream_consumed = False
    self._body: bytes | None = None
    self._json: Any = EMPTY
    self._query_params: dict[str, list[str]] | None = None
    self._headers: dict[str, str] | None = None

query_params property #

query_params: dict[str, list[str]]

headers property #

headers: dict[str, str]

method property #

method: str

stream async #

stream() -> AsyncGenerator[bytes, None]
Source code in faststream/asgi/request.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
async def stream(self) -> AsyncGenerator[bytes, None]:
    if self._body is not None:
        yield self._body
        yield b""
        return
    if self._stream_consumed:
        msg = "Stream consumed"
        raise RuntimeError(msg)
    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":  # pragma: no branch
            self._is_disconnected = True
            raise ClientDisconnectError
    yield b""

body async #

body() -> bytes
Source code in faststream/asgi/request.py 
47
48
49
50
51
async def body(self) -> bytes:
    if self._body is None:
        chunks: list[bytes] = [chunk async for chunk in self.stream()]
        self._body = b"".join(chunks)
    return self._body

json async #

json() -> Any
Source code in faststream/asgi/request.py 
53
54
55
56
57
async def json(self) -> Any:
    if self._json is EMPTY:
        body = await self.body()
        self._json = json.loads(body)
    return self._json