StreamMessage

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

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

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

@classmethod
def get_session(
    cls,
    dependant: Dependant,
    dependency_overrides_provider: Optional[Any] = None,
) -> Callable[[NativeMessage[Any]], Awaitable[SendableMessage]]:
    """Creates a session for handling requests.

    Args:
        dependant: The dependant object representing the session.
        dependency_overrides_provider: Optional provider for dependency overrides.

    Returns:
        A callable that takes a native message and returns an awaitable sendable message.

    Raises:
        AssertionError: If the dependant call is not defined.

    Note:
        This function is used to create a session for handling requests. It takes a dependant object, which represents the session, and a dependency overrides provider, which allows for overriding dependencies. It returns a callable that takes a native message and returns an awaitable sendable message. The session is created based on the dependant object and the message passed to the callable. The session is then used to call the function obtained from the dependant object, and the result is returned.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    assert dependant.call  # nosec B101

    func = get_app(dependant, dependency_overrides_provider)

    dependencies_names = tuple(i.name for i in dependant.dependencies)

    first_arg = next(
        dropwhile(
            lambda i: i in dependencies_names,
            inspect.signature(dependant.call).parameters,
        ),
        None,
    )

    async def app(message: NativeMessage[Any]) -> SendableMessage:
        """An asynchronous function that processes an incoming message and returns a sendable message.

        Args:
            message : The incoming message to be processed

        Returns:
            The sendable message

        Raises:
            TypeError: If the body of the message is not a dictionary
        !!! note

            The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
        """
        body = message.decoded_body
        if first_arg is not None:
            if not isinstance(body, dict) and not isinstance(body, list):
                fastapi_body: Any = {first_arg: body}
            else:
                fastapi_body = body

            session = cls(fastapi_body, message.headers, message.path)
        else:
            session = cls()
        return await func(session)

    return app

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

async def json(self) -> typing.Any:
    if not hasattr(self, "_json"):
        body = await self.body()
        self._json = json.loads(body)
    return self._json

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

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")
    self._stream_consumed = True
    while True:
        message = await self._receive()
        if message["type"] == "http.request":
            body = message.get("body", b"")
            if body:
                yield body
            if not message.get("more_body", False):
                break
        elif message["type"] == "http.disconnect":
            self._is_disconnected = True
            raise ClientDisconnect()
    yield b""

def url_for(self, __name: str, **path_params: typing.Any) -> URL:
    router: Router = self.scope["router"]
    url_path = router.url_path_for(__name, **path_params)
    return url_path.make_absolute_url(base_url=self.base_url)