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

        fastapi_body: Union[AnyDict, List[Any]]
        if first_arg is not None:
            if isinstance(body, dict):
                path = fastapi_body = body or {}
            elif isinstance(body, list):
                fastapi_body, path = body, {}
            else:
                path = fastapi_body = {first_arg: body}

            session = cls(
                body=fastapi_body,
                headers=message.headers,
                path={**path, **message.path},
            )

        else:
            session = cls(
                body={},
                headers={},
                path={},
            )

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

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)