Загрузить файлы в «venv/Lib/site-packages/starlette»

venv/Lib/site-packages/starlette/endpoints.py

@@ -0,0 +1,127 @@
+from __future__ import annotations
+
+import json
+from collections.abc import Callable, Generator
+from typing import Any, Literal
+
+from starlette import status
+from starlette._utils import is_async_callable
+from starlette.concurrency import run_in_threadpool
+from starlette.exceptions import HTTPException
+from starlette.requests import Request
+from starlette.responses import PlainTextResponse, Response
+from starlette.types import Message, Receive, Scope, Send
+from starlette.websockets import WebSocket
+
+
+class HTTPEndpoint:
+    def __init__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        assert scope["type"] == "http"
+        self.scope = scope
+        self.receive = receive
+        self.send = send
+        self._allowed_methods = [
+            method
+            for method in ("GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS")
+            if getattr(self, method.lower(), None) is not None
+        ]
+
+    def __await__(self) -> Generator[Any, None, None]:
+        return self.dispatch().__await__()
+
+    async def dispatch(self) -> None:
+        request = Request(self.scope, receive=self.receive)
+        handler_name = "get" if request.method == "HEAD" and not hasattr(self, "head") else request.method.lower()
+
+        handler: Callable[[Request], Any]
+        if request.method in self._allowed_methods or (request.method == "HEAD" and "GET" in self._allowed_methods):
+            handler = getattr(self, handler_name)
+        else:
+            handler = self.method_not_allowed
+        is_async = is_async_callable(handler)
+        if is_async:
+            response = await handler(request)
+        else:
+            response = await run_in_threadpool(handler, request)
+        await response(self.scope, self.receive, self.send)
+
+    async def method_not_allowed(self, request: Request) -> Response:
+        # If we're running inside a starlette application then raise an
+        # exception, so that the configurable exception handler can deal with
+        # returning the response. For plain ASGI apps, just return the response.
+        headers = {"Allow": ", ".join(self._allowed_methods)}
+        if "app" in self.scope:
+            raise HTTPException(status_code=405, headers=headers)
+        return PlainTextResponse("Method Not Allowed", status_code=405, headers=headers)
+
+
+class WebSocketEndpoint:
+    encoding: Literal["text", "bytes", "json"] | None = None
+
+    def __init__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        assert scope["type"] == "websocket"
+        self.scope = scope
+        self.receive = receive
+        self.send = send
+
+    def __await__(self) -> Generator[Any, None, None]:
+        return self.dispatch().__await__()
+
+    async def dispatch(self) -> None:
+        websocket = WebSocket(self.scope, receive=self.receive, send=self.send)
+        await self.on_connect(websocket)
+
+        close_code = status.WS_1000_NORMAL_CLOSURE
+
+        try:
+            while True:
+                message = await websocket.receive()
+                if message["type"] == "websocket.receive":
+                    data = await self.decode(websocket, message)
+                    await self.on_receive(websocket, data)
+                elif message["type"] == "websocket.disconnect":  # pragma: no branch
+                    close_code = int(message.get("code") or status.WS_1000_NORMAL_CLOSURE)
+                    break
+        except Exception as exc:
+            close_code = status.WS_1011_INTERNAL_ERROR
+            raise exc
+        finally:
+            await self.on_disconnect(websocket, close_code)
+
+    async def decode(self, websocket: WebSocket, message: Message) -> Any:
+        if self.encoding == "text":
+            if "text" not in message:
+                await websocket.close(code=status.WS_1003_UNSUPPORTED_DATA)
+                raise RuntimeError("Expected text websocket messages, but got bytes")
+            return message["text"]
+
+        elif self.encoding == "bytes":
+            if "bytes" not in message:
+                await websocket.close(code=status.WS_1003_UNSUPPORTED_DATA)
+                raise RuntimeError("Expected bytes websocket messages, but got text")
+            return message["bytes"]
+
+        elif self.encoding == "json":
+            if message.get("text") is not None:
+                text = message["text"]
+            else:
+                text = message["bytes"].decode("utf-8")
+
+            try:
+                return json.loads(text)
+            except json.decoder.JSONDecodeError:
+                await websocket.close(code=status.WS_1003_UNSUPPORTED_DATA)
+                raise RuntimeError("Malformed JSON data received.")
+
+        assert self.encoding is None, f"Unsupported 'encoding' attribute {self.encoding}"
+        return message["text"] if message.get("text") else message["bytes"]
+
+    async def on_connect(self, websocket: WebSocket) -> None:
+        """Override to handle an incoming websocket connection"""
+        await websocket.accept()
+
+    async def on_receive(self, websocket: WebSocket, data: Any) -> None:
+        """Override to handle an incoming websocket message"""
+
+    async def on_disconnect(self, websocket: WebSocket, close_code: int) -> None:
+        """Override to handle a disconnecting websocket"""

venv/Lib/site-packages/starlette/exceptions.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import http
+from collections.abc import Mapping
+
+
+class HTTPException(Exception):
+    def __init__(self, status_code: int, detail: str | None = None, headers: Mapping[str, str] | None = None) -> None:
+        if detail is None:
+            detail = http.HTTPStatus(status_code).phrase
+        self.status_code = status_code
+        self.detail = detail
+        self.headers = headers
+
+    def __str__(self) -> str:
+        return f"{self.status_code}: {self.detail}"
+
+    def __repr__(self) -> str:
+        class_name = self.__class__.__name__
+        return f"{class_name}(status_code={self.status_code!r}, detail={self.detail!r})"
+
+
+class WebSocketException(Exception):
+    def __init__(self, code: int, reason: str | None = None) -> None:
+        self.code = code
+        self.reason = reason or ""
+
+    def __str__(self) -> str:
+        return f"{self.code}: {self.reason}"
+
+    def __repr__(self) -> str:
+        class_name = self.__class__.__name__
+        return f"{class_name}(code={self.code!r}, reason={self.reason!r})"
+
+
+class StarletteDeprecationWarning(UserWarning):
+    """A custom deprecation warning for Starlette.
+
+    Unlike the built-in DeprecationWarning, this inherits from UserWarning to ensure it is visible by default, helping
+    users discover deprecated features without needing to enable warnings explicitly.
+
+    Reference: https://sethmlarson.dev/deprecations-via-warnings-dont-work-for-python-libraries
+    """

venv/Lib/site-packages/starlette/formparsers.py

@@ -0,0 +1,297 @@
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+from dataclasses import dataclass, field
+from enum import Enum
+from tempfile import SpooledTemporaryFile
+from typing import TYPE_CHECKING
+from urllib.parse import unquote_plus
+
+from starlette.datastructures import FormData, Headers, UploadFile
+
+if TYPE_CHECKING:
+    import python_multipart as multipart
+    from python_multipart.multipart import MultipartCallbacks, QuerystringCallbacks, parse_options_header
+else:
+    try:
+        try:
+            import python_multipart as multipart
+            from python_multipart.multipart import parse_options_header
+        except ModuleNotFoundError:  # pragma: no cover
+            import multipart
+            from multipart.multipart import parse_options_header
+    except ModuleNotFoundError:  # pragma: no cover
+        multipart = None
+        parse_options_header = None
+
+
+class FormMessage(Enum):
+    FIELD_START = 1
+    FIELD_NAME = 2
+    FIELD_DATA = 3
+    FIELD_END = 4
+    END = 5
+
+
+@dataclass
+class MultipartPart:
+    content_disposition: bytes | None = None
+    field_name: str = ""
+    data: bytearray = field(default_factory=bytearray)
+    file: UploadFile | None = None
+    item_headers: list[tuple[bytes, bytes]] = field(default_factory=list)
+
+
+def _user_safe_decode(src: bytes | bytearray, codec: str) -> str:
+    try:
+        return src.decode(codec)
+    except (UnicodeDecodeError, LookupError):
+        return src.decode("latin-1")
+
+
+class MultiPartException(Exception):
+    def __init__(self, message: str) -> None:
+        self.message = message
+
+
+class FormParser:
+    def __init__(
+        self,
+        headers: Headers,
+        stream: AsyncGenerator[bytes, None],
+        *,
+        max_fields: int | float = 1000,
+        max_part_size: int = 1024 * 1024,  # 1MB
+    ) -> None:
+        assert multipart is not None, "The `python-multipart` library must be installed to use form parsing."
+        self.headers = headers
+        self.stream = stream
+        self.max_fields = max_fields
+        self.max_part_size = max_part_size
+        self.messages: list[tuple[FormMessage, bytes]] = []
+        self._current_field_size = 0
+        self._current_fields = 0
+
+    def on_field_start(self) -> None:
+        self._current_field_size = 0
+        message = (FormMessage.FIELD_START, b"")
+        self.messages.append(message)
+
+    def on_field_name(self, data: bytes, start: int, end: int) -> None:
+        self._current_field_size += end - start
+        if self._current_field_size > self.max_part_size:
+            raise MultiPartException(f"Field exceeded maximum size of {int(self.max_part_size / 1024)}KB.")
+        message = (FormMessage.FIELD_NAME, data[start:end])
+        self.messages.append(message)
+
+    def on_field_data(self, data: bytes, start: int, end: int) -> None:
+        self._current_field_size += end - start
+        if self._current_field_size > self.max_part_size:
+            raise MultiPartException(f"Field exceeded maximum size of {int(self.max_part_size / 1024)}KB.")
+        message = (FormMessage.FIELD_DATA, data[start:end])
+        self.messages.append(message)
+
+    def on_field_end(self) -> None:
+        self._current_fields += 1
+        if self._current_fields > self.max_fields:
+            raise MultiPartException(f"Too many fields. Maximum number of fields is {self.max_fields}.")
+        message = (FormMessage.FIELD_END, b"")
+        self.messages.append(message)
+
+    def on_end(self) -> None:
+        message = (FormMessage.END, b"")
+        self.messages.append(message)
+
+    async def parse(self) -> FormData:
+        # Callbacks dictionary.
+        callbacks: QuerystringCallbacks = {
+            "on_field_start": self.on_field_start,
+            "on_field_name": self.on_field_name,
+            "on_field_data": self.on_field_data,
+            "on_field_end": self.on_field_end,
+            "on_end": self.on_end,
+        }
+
+        # Create the parser.
+        parser = multipart.QuerystringParser(callbacks)
+        field_name = bytearray()
+        field_value = bytearray()
+
+        items: list[tuple[str, str | UploadFile]] = []
+
+        # Feed the parser with data from the request.
+        async for chunk in self.stream:
+            if chunk:
+                parser.write(chunk)
+            else:
+                parser.finalize()
+            messages = list(self.messages)
+            self.messages.clear()
+            for message_type, message_bytes in messages:
+                if message_type == FormMessage.FIELD_START:
+                    field_name = bytearray()
+                    field_value = bytearray()
+                elif message_type == FormMessage.FIELD_NAME:
+                    field_name.extend(message_bytes)
+                elif message_type == FormMessage.FIELD_DATA:
+                    field_value.extend(message_bytes)
+                elif message_type == FormMessage.FIELD_END:
+                    name = unquote_plus(field_name.decode("latin-1"))
+                    value = unquote_plus(field_value.decode("latin-1"))
+                    items.append((name, value))
+
+        return FormData(items)
+
+
+class MultiPartParser:
+    spool_max_size = 1024 * 1024  # 1MB
+    """The maximum size of the spooled temporary file used to store file data."""
+    max_part_size = 1024 * 1024  # 1MB
+    """The maximum size of a part in the multipart request."""
+
+    def __init__(
+        self,
+        headers: Headers,
+        stream: AsyncGenerator[bytes, None],
+        *,
+        max_files: int | float = 1000,
+        max_fields: int | float = 1000,
+        max_part_size: int = 1024 * 1024,  # 1MB
+    ) -> None:
+        assert multipart is not None, "The `python-multipart` library must be installed to use form parsing."
+        self.headers = headers
+        self.stream = stream
+        self.max_files = max_files
+        self.max_fields = max_fields
+        self.items: list[tuple[str, str | UploadFile]] = []
+        self._current_files = 0
+        self._current_fields = 0
+        self._current_partial_header_name: bytes = b""
+        self._current_partial_header_value: bytes = b""
+        self._current_part = MultipartPart()
+        self._charset = ""
+        self._file_parts_to_write: list[tuple[MultipartPart, bytes]] = []
+        self._file_parts_to_finish: list[MultipartPart] = []
+        self._files_to_close_on_error: list[SpooledTemporaryFile[bytes]] = []
+        self.max_part_size = max_part_size
+
+    def on_part_begin(self) -> None:
+        self._current_part = MultipartPart()
+
+    def on_part_data(self, data: bytes, start: int, end: int) -> None:
+        message_bytes = data[start:end]
+        if self._current_part.file is None:
+            if len(self._current_part.data) + len(message_bytes) > self.max_part_size:
+                raise MultiPartException(f"Part exceeded maximum size of {int(self.max_part_size / 1024)}KB.")
+            self._current_part.data.extend(message_bytes)
+        else:
+            self._file_parts_to_write.append((self._current_part, message_bytes))
+
+    def on_part_end(self) -> None:
+        if self._current_part.file is None:
+            self.items.append(
+                (
+                    self._current_part.field_name,
+                    _user_safe_decode(self._current_part.data, self._charset),
+                )
+            )
+        else:
+            self._file_parts_to_finish.append(self._current_part)
+            # The file can be added to the items right now even though it's not
+            # finished yet, because it will be finished in the `parse()` method, before
+            # self.items is used in the return value.
+            self.items.append((self._current_part.field_name, self._current_part.file))
+
+    def on_header_field(self, data: bytes, start: int, end: int) -> None:
+        self._current_partial_header_name += data[start:end]
+
+    def on_header_value(self, data: bytes, start: int, end: int) -> None:
+        self._current_partial_header_value += data[start:end]
+
+    def on_header_end(self) -> None:
+        field = self._current_partial_header_name.lower()
+        if field == b"content-disposition":
+            self._current_part.content_disposition = self._current_partial_header_value
+        self._current_part.item_headers.append((field, self._current_partial_header_value))
+        self._current_partial_header_name = b""
+        self._current_partial_header_value = b""
+
+    def on_headers_finished(self) -> None:
+        disposition, options = parse_options_header(self._current_part.content_disposition)
+        try:
+            self._current_part.field_name = _user_safe_decode(options[b"name"], self._charset)
+        except KeyError:
+            raise MultiPartException('The Content-Disposition header field "name" must be provided.')
+        if b"filename" in options:
+            self._current_files += 1
+            if self._current_files > self.max_files:
+                raise MultiPartException(f"Too many files. Maximum number of files is {self.max_files}.")
+            filename = _user_safe_decode(options[b"filename"], self._charset)
+            tempfile = SpooledTemporaryFile(max_size=self.spool_max_size)
+            self._files_to_close_on_error.append(tempfile)
+            self._current_part.file = UploadFile(
+                file=tempfile,  # type: ignore[arg-type]
+                size=0,
+                filename=filename,
+                headers=Headers(raw=self._current_part.item_headers),
+            )
+        else:
+            self._current_fields += 1
+            if self._current_fields > self.max_fields:
+                raise MultiPartException(f"Too many fields. Maximum number of fields is {self.max_fields}.")
+            self._current_part.file = None
+
+    def on_end(self) -> None:
+        pass
+
+    async def parse(self) -> FormData:
+        # Parse the Content-Type header to get the multipart boundary.
+        _, params = parse_options_header(self.headers["Content-Type"])
+        charset = params.get(b"charset", "utf-8")
+        if isinstance(charset, bytes):
+            charset = charset.decode("latin-1")
+        self._charset = charset
+        try:
+            boundary = params[b"boundary"]
+        except KeyError:
+            raise MultiPartException("Missing boundary in multipart.")
+
+        # Callbacks dictionary.
+        callbacks: MultipartCallbacks = {
+            "on_part_begin": self.on_part_begin,
+            "on_part_data": self.on_part_data,
+            "on_part_end": self.on_part_end,
+            "on_header_field": self.on_header_field,
+            "on_header_value": self.on_header_value,
+            "on_header_end": self.on_header_end,
+            "on_headers_finished": self.on_headers_finished,
+            "on_end": self.on_end,
+        }
+
+        # Create the parser.
+        parser = multipart.MultipartParser(boundary, callbacks)
+        try:
+            # Feed the parser with data from the request.
+            async for chunk in self.stream:
+                parser.write(chunk)
+                # Write file data, it needs to use await with the UploadFile methods
+                # that call the corresponding file methods *in a threadpool*,
+                # otherwise, if they were called directly in the callback methods above
+                # (regular, non-async functions), that would block the event loop in
+                # the main thread.
+                for part, data in self._file_parts_to_write:
+                    assert part.file  # for type checkers
+                    await part.file.write(data)
+                for part in self._file_parts_to_finish:
+                    assert part.file  # for type checkers
+                    await part.file.seek(0)
+                self._file_parts_to_write.clear()
+                self._file_parts_to_finish.clear()
+            parser.finalize()
+        except (MultiPartException, OSError) as exc:
+            # Close all the files if there was an error.
+            for file in self._files_to_close_on_error:
+                file.close()
+            raise exc
+
+        return FormData(self.items)

venv/Lib/site-packages/starlette/requests.py

@@ -0,0 +1,348 @@
+from __future__ import annotations
+
+import json
+import sys
+from collections.abc import AsyncGenerator, Iterator, Mapping
+from http import cookies as http_cookies
+from typing import TYPE_CHECKING, Any, Generic, NoReturn, cast
+
+import anyio
+
+from starlette._utils import AwaitableOrContextManager, AwaitableOrContextManagerWrapper
+from starlette.datastructures import URL, Address, FormData, Headers, QueryParams, State
+from starlette.exceptions import HTTPException
+from starlette.formparsers import FormParser, MultiPartException, MultiPartParser
+from starlette.types import Message, Receive, Scope, Send
+
+if TYPE_CHECKING:
+    from python_multipart.multipart import parse_options_header
+
+    from starlette.applications import Starlette
+    from starlette.middleware.sessions import Session
+    from starlette.routing import Router
+else:
+    try:
+        try:
+            from python_multipart.multipart import parse_options_header
+        except ModuleNotFoundError:  # pragma: no cover
+            from multipart.multipart import parse_options_header
+    except ModuleNotFoundError:  # pragma: no cover
+        parse_options_header = None
+
+if sys.version_info >= (3, 13):  # pragma: no cover
+    from typing import TypeVar
+else:  # pragma: no cover
+    from typing_extensions import TypeVar
+
+SERVER_PUSH_HEADERS_TO_COPY = {
+    "accept",
+    "accept-encoding",
+    "accept-language",
+    "cache-control",
+    "user-agent",
+}
+
+
+def cookie_parser(cookie_string: str) -> dict[str, str]:
+    """
+    This function parses a ``Cookie`` HTTP header into a dict of key/value pairs.
+
+    It attempts to mimic browser cookie parsing behavior: browsers and web servers
+    frequently disregard the spec (RFC 6265) when setting and reading cookies,
+    so we attempt to suit the common scenarios here.
+
+    This function has been adapted from Django 3.1.0.
+    Note: we are explicitly _NOT_ using `SimpleCookie.load` because it is based
+    on an outdated spec and will fail on lots of input we want to support
+    """
+    cookie_dict: dict[str, str] = {}
+    for chunk in cookie_string.split(";"):
+        if "=" in chunk:
+            key, val = chunk.split("=", 1)
+        else:
+            # Assume an empty name per
+            # https://bugzilla.mozilla.org/show_bug.cgi?id=169091
+            key, val = "", chunk
+        key, val = key.strip(), val.strip()
+        if key or val:
+            # unquote using Python's algorithm.
+            cookie_dict[key] = http_cookies._unquote(val)
+    return cookie_dict
+
+
+class ClientDisconnect(Exception):
+    pass
+
+
+StateT = TypeVar("StateT", bound=Mapping[str, Any] | State, default=State)
+
+
+class HTTPConnection(Mapping[str, Any], Generic[StateT]):
+    """
+    A base class for incoming HTTP connections, that is used to provide
+    any functionality that is common to both `Request` and `WebSocket`.
+    """
+
+    def __init__(self, scope: Scope, receive: Receive | None = None) -> None:
+        assert scope["type"] in ("http", "websocket")
+        self.scope = scope
+
+    def __getitem__(self, key: str) -> Any:
+        return self.scope[key]
+
+    def __iter__(self) -> Iterator[str]:
+        return iter(self.scope)
+
+    def __len__(self) -> int:
+        return len(self.scope)
+
+    # Don't use the `abc.Mapping.__eq__` implementation.
+    # Connection instances should never be considered equal
+    # unless `self is other`.
+    __eq__ = object.__eq__
+    __hash__ = object.__hash__
+
+    @property
+    def app(self) -> Any:
+        return self.scope["app"]
+
+    @property
+    def url(self) -> URL:
+        if not hasattr(self, "_url"):  # pragma: no branch
+            self._url = URL(scope=self.scope)
+        return self._url
+
+    @property
+    def base_url(self) -> URL:
+        if not hasattr(self, "_base_url"):
+            base_url_scope = dict(self.scope)
+            # This is used by request.url_for, it might be used inside a Mount which
+            # would have its own child scope with its own root_path, but the base URL
+            # for url_for should still be the top level app root path.
+            app_root_path = base_url_scope.get("app_root_path", base_url_scope.get("root_path", ""))
+            path = app_root_path
+            if not path.endswith("/"):
+                path += "/"
+            base_url_scope["path"] = path
+            base_url_scope["query_string"] = b""
+            base_url_scope["root_path"] = app_root_path
+            self._base_url = URL(scope=base_url_scope)
+        return self._base_url
+
+    @property
+    def headers(self) -> Headers:
+        if not hasattr(self, "_headers"):
+            self._headers = Headers(scope=self.scope)
+        return self._headers
+
+    @property
+    def query_params(self) -> QueryParams:
+        if not hasattr(self, "_query_params"):  # pragma: no branch
+            self._query_params = QueryParams(self.scope["query_string"])
+        return self._query_params
+
+    @property
+    def path_params(self) -> dict[str, Any]:
+        path_params: dict[str, Any] = self.scope.get("path_params", {})
+        return path_params
+
+    @property
+    def cookies(self) -> dict[str, str]:
+        if not hasattr(self, "_cookies"):
+            cookies: dict[str, str] = {}
+            cookie_headers = self.headers.getlist("cookie")
+
+            for header in cookie_headers:
+                cookies.update(cookie_parser(header))
+
+            self._cookies = cookies
+        return self._cookies
+
+    @property
+    def client(self) -> Address | None:
+        # client is a 2 item tuple of (host, port), None if missing
+        host_port = self.scope.get("client")
+        if host_port is not None:
+            return Address(*host_port)
+        return None
+
+    @property
+    def session(self) -> dict[str, Any]:
+        assert "session" in self.scope, "SessionMiddleware must be installed to access request.session"
+        session: Session = self.scope["session"]
+        # We keep the hasattr in case people actually use their own `SessionMiddleware` implementation.
+        if hasattr(session, "mark_accessed"):  # pragma: no branch
+            session.mark_accessed()
+        return session
+
+    @property
+    def auth(self) -> Any:
+        assert "auth" in self.scope, "AuthenticationMiddleware must be installed to access request.auth"
+        return self.scope["auth"]
+
+    @property
+    def user(self) -> Any:
+        assert "user" in self.scope, "AuthenticationMiddleware must be installed to access request.user"
+        return self.scope["user"]
+
+    @property
+    def state(self) -> StateT:
+        if not hasattr(self, "_state"):
+            # Ensure 'state' has an empty dict if it's not already populated.
+            self.scope.setdefault("state", {})
+            # Create a state instance with a reference to the dict in which it should
+            # store info
+            self._state = State(self.scope["state"])
+        return cast(StateT, self._state)
+
+    def url_for(self, name: str, /, **path_params: Any) -> URL:
+        url_path_provider: Router | Starlette | None = self.scope.get("router") or self.scope.get("app")
+        if url_path_provider is None:
+            raise RuntimeError("The `url_for` method can only be used inside a Starlette application or with a router.")
+        url_path = url_path_provider.url_path_for(name, **path_params)
+        return url_path.make_absolute_url(base_url=self.base_url)
+
+
+async def empty_receive() -> NoReturn:
+    raise RuntimeError("Receive channel has not been made available")
+
+
+async def empty_send(message: Message) -> NoReturn:
+    raise RuntimeError("Send channel has not been made available")
+
+
+class Request(HTTPConnection[StateT]):
+    _form: FormData | None
+
+    def __init__(self, scope: Scope, receive: Receive = empty_receive, send: Send = empty_send):
+        super().__init__(scope)
+        assert scope["type"] == "http"
+        self._receive = receive
+        self._send = send
+        self._stream_consumed = False
+        self._is_disconnected = False
+        self._form = None
+
+    @property
+    def method(self) -> str:
+        return cast(str, self.scope["method"])
+
+    @property
+    def receive(self) -> Receive:
+        return self._receive
+
+    async def stream(self) -> 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":  # pragma: no branch
+                self._is_disconnected = True
+                raise ClientDisconnect()
+        yield b""
+
+    async def body(self) -> bytes:
+        if not hasattr(self, "_body"):
+            chunks: list[bytes] = []
+            async for chunk in self.stream():
+                chunks.append(chunk)
+            self._body = b"".join(chunks)
+        return self._body
+
+    async def json(self) -> Any:
+        if not hasattr(self, "_json"):  # pragma: no branch
+            body = await self.body()
+            self._json = json.loads(body)
+        return self._json
+
+    async def _get_form(
+        self,
+        *,
+        max_files: int | float = 1000,
+        max_fields: int | float = 1000,
+        max_part_size: int = 1024 * 1024,
+    ) -> FormData:
+        if self._form is None:  # pragma: no branch
+            assert parse_options_header is not None, (
+                "The `python-multipart` library must be installed to use form parsing."
+            )
+            content_type_header = self.headers.get("Content-Type")
+            content_type: bytes
+            content_type, _ = parse_options_header(content_type_header)
+            if content_type == b"multipart/form-data":
+                try:
+                    multipart_parser = MultiPartParser(
+                        self.headers,
+                        self.stream(),
+                        max_files=max_files,
+                        max_fields=max_fields,
+                        max_part_size=max_part_size,
+                    )
+                    self._form = await multipart_parser.parse()
+                except MultiPartException as exc:
+                    if "app" in self.scope:
+                        raise HTTPException(status_code=400, detail=exc.message)
+                    raise exc
+            elif content_type == b"application/x-www-form-urlencoded":
+                try:
+                    form_parser = FormParser(
+                        self.headers,
+                        self.stream(),
+                        max_fields=max_fields,
+                        max_part_size=max_part_size,
+                    )
+                    self._form = await form_parser.parse()
+                except MultiPartException as exc:
+                    if "app" in self.scope:
+                        raise HTTPException(status_code=400, detail=exc.message)
+                    raise exc
+            else:
+                self._form = FormData()
+        return self._form
+
+    def form(
+        self,
+        *,
+        max_files: int | float = 1000,
+        max_fields: int | float = 1000,
+        max_part_size: int = 1024 * 1024,
+    ) -> AwaitableOrContextManager[FormData]:
+        return AwaitableOrContextManagerWrapper(
+            self._get_form(max_files=max_files, max_fields=max_fields, max_part_size=max_part_size)
+        )
+
+    async def close(self) -> None:
+        if self._form is not None:  # pragma: no branch
+            await self._form.close()
+
+    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 send_push_promise(self, path: str) -> None:
+        if "http.response.push" in self.scope.get("extensions", {}):
+            raw_headers: list[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})