From 40151d2f9f0a8f248d3ddce6e053eacad01edcac Mon Sep 17 00:00:00 2001 From: Polina Date: Thu, 2 Jul 2026 20:48:00 +0000 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=B8?= =?UTF-8?q?=D1=82=D1=8C=20=D1=84=D0=B0=D0=B9=D0=BB=D1=8B=20=D0=B2=20=C2=AB?= =?UTF-8?q?venv/Lib/site-packages/starlette=C2=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Lib/site-packages/starlette/background.py | 36 + .../site-packages/starlette/concurrency.py | 59 ++ venv/Lib/site-packages/starlette/config.py | 140 ++++ .../Lib/site-packages/starlette/convertors.py | 89 +++ .../site-packages/starlette/datastructures.py | 704 ++++++++++++++++++ 5 files changed, 1028 insertions(+) create mode 100644 venv/Lib/site-packages/starlette/background.py create mode 100644 venv/Lib/site-packages/starlette/concurrency.py create mode 100644 venv/Lib/site-packages/starlette/config.py create mode 100644 venv/Lib/site-packages/starlette/convertors.py create mode 100644 venv/Lib/site-packages/starlette/datastructures.py diff --git a/venv/Lib/site-packages/starlette/background.py b/venv/Lib/site-packages/starlette/background.py new file mode 100644 index 0000000..55135ee --- /dev/null +++ b/venv/Lib/site-packages/starlette/background.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from collections.abc import Callable, Sequence +from typing import Any, ParamSpec + +from starlette._utils import is_async_callable +from starlette.concurrency import run_in_threadpool + +P = ParamSpec("P") + + +class BackgroundTask: + def __init__(self, func: Callable[P, Any], *args: P.args, **kwargs: P.kwargs) -> None: + self.func = func + self.args = args + self.kwargs = kwargs + self.is_async = is_async_callable(func) + + async def __call__(self) -> None: + if self.is_async: + await self.func(*self.args, **self.kwargs) + else: + await run_in_threadpool(self.func, *self.args, **self.kwargs) + + +class BackgroundTasks(BackgroundTask): + def __init__(self, tasks: Sequence[BackgroundTask] | None = None): + self.tasks = list(tasks) if tasks else [] + + def add_task(self, func: Callable[P, Any], *args: P.args, **kwargs: P.kwargs) -> None: + task = BackgroundTask(func, *args, **kwargs) + self.tasks.append(task) + + async def __call__(self) -> None: + for task in self.tasks: + await task() diff --git a/venv/Lib/site-packages/starlette/concurrency.py b/venv/Lib/site-packages/starlette/concurrency.py new file mode 100644 index 0000000..a2c55be --- /dev/null +++ b/venv/Lib/site-packages/starlette/concurrency.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +import functools +import warnings +from collections.abc import AsyncIterator, Callable, Coroutine, Iterable, Iterator +from typing import ParamSpec, TypeVar + +import anyio.to_thread + +from starlette.exceptions import StarletteDeprecationWarning + +P = ParamSpec("P") +T = TypeVar("T") + + +async def run_until_first_complete(*args: tuple[Callable, dict]) -> None: # type: ignore[type-arg] + warnings.warn( + "run_until_first_complete is deprecated and will be removed in a future version.", + StarletteDeprecationWarning, + ) + + async with anyio.create_task_group() as task_group: + + async def run(func: Callable[[], Coroutine]) -> None: # type: ignore[type-arg] + await func() + task_group.cancel_scope.cancel() + + for func, kwargs in args: + task_group.start_soon(run, functools.partial(func, **kwargs)) + + +async def run_in_threadpool(func: Callable[P, T], *args: P.args, **kwargs: P.kwargs) -> T: + func = functools.partial(func, *args, **kwargs) + return await anyio.to_thread.run_sync(func) + + +class _StopIteration(Exception): + pass + + +def _next(iterator: Iterator[T]) -> T: + # We can't raise `StopIteration` from within the threadpool iterator + # and catch it outside that context, so we coerce them into a different + # exception type. + try: + return next(iterator) + except StopIteration: + raise _StopIteration + + +async def iterate_in_threadpool( + iterator: Iterable[T], +) -> AsyncIterator[T]: + as_iterator = iter(iterator) + while True: + try: + yield await anyio.to_thread.run_sync(_next, as_iterator) + except _StopIteration: + break diff --git a/venv/Lib/site-packages/starlette/config.py b/venv/Lib/site-packages/starlette/config.py new file mode 100644 index 0000000..56c25ca --- /dev/null +++ b/venv/Lib/site-packages/starlette/config.py @@ -0,0 +1,140 @@ +from __future__ import annotations + +import os +import warnings +from collections.abc import Callable, Iterator, Mapping, MutableMapping +from pathlib import Path +from typing import Any, TypeVar, overload + + +class undefined: + pass + + +class EnvironError(Exception): + pass + + +class Environ(MutableMapping[str, str]): + def __init__(self, environ: MutableMapping[str, str] = os.environ): + self._environ = environ + self._has_been_read: set[str] = set() + + def __getitem__(self, key: str) -> str: + self._has_been_read.add(key) + return self._environ.__getitem__(key) + + def __setitem__(self, key: str, value: str) -> None: + if key in self._has_been_read: + raise EnvironError(f"Attempting to set environ['{key}'], but the value has already been read.") + self._environ.__setitem__(key, value) + + def __delitem__(self, key: str) -> None: + if key in self._has_been_read: + raise EnvironError(f"Attempting to delete environ['{key}'], but the value has already been read.") + self._environ.__delitem__(key) + + def __iter__(self) -> Iterator[str]: + return iter(self._environ) + + def __len__(self) -> int: + return len(self._environ) + + +environ = Environ() + +T = TypeVar("T") + + +class Config: + def __init__( + self, + env_file: str | Path | None = None, + environ: Mapping[str, str] = environ, + env_prefix: str = "", + encoding: str = "utf-8", + ) -> None: + self.environ = environ + self.env_prefix = env_prefix + self.file_values: dict[str, str] = {} + if env_file is not None: + if not os.path.isfile(env_file): + warnings.warn(f"Config file '{env_file}' not found.") + else: + self.file_values = self._read_file(env_file, encoding) + + @overload + def __call__(self, key: str, *, default: None) -> str | None: ... + + @overload + def __call__(self, key: str, cast: type[T], default: T = ...) -> T: ... + + @overload + def __call__(self, key: str, cast: type[str] = ..., default: str = ...) -> str: ... + + @overload + def __call__( + self, + key: str, + cast: Callable[[Any], T] = ..., + default: Any = ..., + ) -> T: ... + + @overload + def __call__(self, key: str, cast: type[str] = ..., default: T = ...) -> T | str: ... + + def __call__( + self, + key: str, + cast: Callable[[Any], Any] | None = None, + default: Any = undefined, + ) -> Any: + return self.get(key, cast, default) + + def get( + self, + key: str, + cast: Callable[[Any], Any] | None = None, + default: Any = undefined, + ) -> Any: + key = self.env_prefix + key + if key in self.environ: + value = self.environ[key] + return self._perform_cast(key, value, cast) + if key in self.file_values: + value = self.file_values[key] + return self._perform_cast(key, value, cast) + if default is not undefined: + return self._perform_cast(key, default, cast) + raise KeyError(f"Config '{key}' is missing, and has no default.") + + def _read_file(self, file_name: str | Path, encoding: str) -> dict[str, str]: + file_values: dict[str, str] = {} + with open(file_name, encoding=encoding) as input_file: + for line in input_file.readlines(): + line = line.strip() + if "=" in line and not line.startswith("#"): + key, value = line.split("=", 1) + key = key.strip() + value = value.strip().strip("\"'") + file_values[key] = value + return file_values + + def _perform_cast( + self, + key: str, + value: Any, + cast: Callable[[Any], Any] | None = None, + ) -> Any: + if cast is None or value is None: + return value + elif cast is bool and isinstance(value, str): + mapping = {"true": True, "1": True, "false": False, "0": False} + value = value.lower() + if value not in mapping: + raise ValueError(f"Config '{key}' has value '{value}'. Not a valid bool.") + return mapping[value] + try: + return cast(value) + except (TypeError, ValueError): + raise ValueError(f"Config '{key}' has value '{value}'. Not a valid {cast.__name__}.") diff --git a/venv/Lib/site-packages/starlette/convertors.py b/venv/Lib/site-packages/starlette/convertors.py new file mode 100644 index 0000000..72b1cf9 --- /dev/null +++ b/venv/Lib/site-packages/starlette/convertors.py @@ -0,0 +1,89 @@ +from __future__ import annotations + +import math +import uuid +from typing import Any, ClassVar, Generic, TypeVar + +T = TypeVar("T") + + +class Convertor(Generic[T]): + regex: ClassVar[str] = "" + + def convert(self, value: str) -> T: + raise NotImplementedError() # pragma: no cover + + def to_string(self, value: T) -> str: + raise NotImplementedError() # pragma: no cover + + +class StringConvertor(Convertor[str]): + regex = "[^/]+" + + def convert(self, value: str) -> str: + return value + + def to_string(self, value: str) -> str: + value = str(value) + assert "/" not in value, "May not contain path separators" + assert value, "Must not be empty" + return value + + +class PathConvertor(Convertor[str]): + regex = ".*" + + def convert(self, value: str) -> str: + return str(value) + + def to_string(self, value: str) -> str: + return str(value) + + +class IntegerConvertor(Convertor[int]): + regex = "[0-9]+" + + def convert(self, value: str) -> int: + return int(value) + + def to_string(self, value: int) -> str: + value = int(value) + assert value >= 0, "Negative integers are not supported" + return str(value) + + +class FloatConvertor(Convertor[float]): + regex = r"[0-9]+(\.[0-9]+)?" + + def convert(self, value: str) -> float: + return float(value) + + def to_string(self, value: float) -> str: + value = float(value) + assert value >= 0.0, "Negative floats are not supported" + assert not math.isnan(value), "NaN values are not supported" + assert not math.isinf(value), "Infinite values are not supported" + return ("%0.20f" % value).rstrip("0").rstrip(".") + + +class UUIDConvertor(Convertor[uuid.UUID]): + regex = "[0-9a-fA-F]{8}-?[0-9a-fA-F]{4}-?[0-9a-fA-F]{4}-?[0-9a-fA-F]{4}-?[0-9a-fA-F]{12}" + + def convert(self, value: str) -> uuid.UUID: + return uuid.UUID(value) + + def to_string(self, value: uuid.UUID) -> str: + return str(value) + + +CONVERTOR_TYPES: dict[str, Convertor[Any]] = { + "str": StringConvertor(), + "path": PathConvertor(), + "int": IntegerConvertor(), + "float": FloatConvertor(), + "uuid": UUIDConvertor(), +} + + +def register_url_convertor(key: str, convertor: Convertor[Any]) -> None: + CONVERTOR_TYPES[key] = convertor diff --git a/venv/Lib/site-packages/starlette/datastructures.py b/venv/Lib/site-packages/starlette/datastructures.py new file mode 100644 index 0000000..33ecb1b --- /dev/null +++ b/venv/Lib/site-packages/starlette/datastructures.py @@ -0,0 +1,704 @@ +from __future__ import annotations + +import re +from collections.abc import ItemsView, Iterable, Iterator, KeysView, Mapping, MutableMapping, Sequence, ValuesView +from shlex import shlex +from typing import Any, BinaryIO, Literal, NamedTuple, TypeVar, cast +from urllib.parse import SplitResult, parse_qsl, urlencode, urlsplit + +from starlette.concurrency import run_in_threadpool +from starlette.types import Scope + + +class Address(NamedTuple): + host: str + port: int + + +_KeyType = TypeVar("_KeyType") +# Mapping keys are invariant but their values are covariant since +# you can only read them +# that is, you can't do `Mapping[str, Animal]()["fido"] = Dog()` +_CovariantValueType = TypeVar("_CovariantValueType", covariant=True) + +# Rejects Host header chars (/, ?, #, @, ...) that would let urlsplit produce a path differing from scope["path"]. +_HOST_RE = re.compile(r"^([a-z0-9.-]+|\[[a-f0-9]*:[a-f0-9.:]+\])(?::[0-9]+)?$", re.IGNORECASE) + + +class URL: + def __init__( + self, + url: str = "", + scope: Scope | None = None, + **components: Any, + ) -> None: + if scope is not None: + assert not url, 'Cannot set both "url" and "scope".' + assert not components, 'Cannot set both "scope" and "**components".' + scheme = scope.get("scheme", "http") + server = scope.get("server", None) + path = scope["path"] + query_string = scope.get("query_string", b"") + + host_header = None + for key, value in scope["headers"]: + if key == b"host": + host_header = value.decode("latin-1") + break + + if host_header is not None and _HOST_RE.fullmatch(host_header): + netloc = host_header + elif server is not None: + host, port = server + default_port = {"http": 80, "https": 443, "ws": 80, "wss": 443}[scheme] + netloc = host if port == default_port else f"{host}:{port}" + else: + netloc = None + + query = query_string.decode() + if netloc is not None: + url = SplitResult(scheme=scheme, netloc=netloc, path=path, query=query, fragment="").geturl() + else: + url = f"{path}?{query}" if query else path + elif components: + assert not url, 'Cannot set both "url" and "**components".' + url = URL("").replace(**components).components.geturl() + + self._url = url + + @property + def components(self) -> SplitResult: + if not hasattr(self, "_components"): + self._components = urlsplit(self._url) + return self._components + + @property + def scheme(self) -> str: + return self.components.scheme + + @property + def netloc(self) -> str: + return self.components.netloc + + @property + def path(self) -> str: + return self.components.path + + @property + def query(self) -> str: + return self.components.query + + @property + def fragment(self) -> str: + return self.components.fragment + + @property + def username(self) -> None | str: + return self.components.username + + @property + def password(self) -> None | str: + return self.components.password + + @property + def hostname(self) -> None | str: + return self.components.hostname + + @property + def port(self) -> int | None: + return self.components.port + + @property + def is_secure(self) -> bool: + return self.scheme in ("https", "wss") + + def replace(self, **kwargs: Any) -> URL: + if "username" in kwargs or "password" in kwargs or "hostname" in kwargs or "port" in kwargs: + hostname = kwargs.pop("hostname", None) + port = kwargs.pop("port", self.port) + username = kwargs.pop("username", self.username) + password = kwargs.pop("password", self.password) + + if hostname is None: + netloc = self.netloc + _, _, hostname = netloc.rpartition("@") + + if hostname and hostname[-1] != "]": + hostname = hostname.rsplit(":", 1)[0] + + netloc = hostname + if port is not None: + netloc += f":{port}" + if username is not None: + userpass = username + if password is not None: + userpass += f":{password}" + netloc = f"{userpass}@{netloc}" + + kwargs["netloc"] = netloc + + components = self.components._replace(**kwargs) + return self.__class__(components.geturl()) + + def include_query_params(self, **kwargs: Any) -> URL: + params = MultiDict(parse_qsl(self.query, keep_blank_values=True)) + params.update({str(key): str(value) for key, value in kwargs.items()}) + query = urlencode(params.multi_items()) + return self.replace(query=query) + + def replace_query_params(self, **kwargs: Any) -> URL: + query = urlencode([(str(key), str(value)) for key, value in kwargs.items()]) + return self.replace(query=query) + + def remove_query_params(self, keys: str | Sequence[str]) -> URL: + if isinstance(keys, str): + keys = [keys] + params = MultiDict(parse_qsl(self.query, keep_blank_values=True)) + for key in keys: + params.pop(key, None) + query = urlencode(params.multi_items()) + return self.replace(query=query) + + def __eq__(self, other: Any) -> bool: + return str(self) == str(other) + + def __str__(self) -> str: + return self._url + + def __repr__(self) -> str: + url = str(self) + if self.password: + url = str(self.replace(password="********")) + return f"{self.__class__.__name__}({repr(url)})" + + +class URLPath(str): + """ + A URL path string that may also hold an associated protocol and/or host. + Used by the routing to return `url_path_for` matches. + """ + + def __new__(cls, path: str, protocol: Literal["http", "websocket", ""] = "", host: str = "") -> URLPath: + assert protocol in ("http", "websocket", "") + return str.__new__(cls, path) + + def __init__(self, path: str, protocol: Literal["http", "websocket", ""] = "", host: str = "") -> None: + self.protocol = protocol + self.host = host + + def make_absolute_url(self, base_url: str | URL) -> URL: + if isinstance(base_url, str): + base_url = URL(base_url) + if self.protocol: + scheme = { + "http": {True: "https", False: "http"}, + "websocket": {True: "wss", False: "ws"}, + }[self.protocol][base_url.is_secure] + else: + scheme = base_url.scheme + + netloc = self.host or base_url.netloc + path = base_url.path.rstrip("/") + str(self) + return URL(scheme=scheme, netloc=netloc, path=path) + + +class Secret: + """ + Holds a string value that should not be revealed in tracebacks etc. + You should cast the value to `str` at the point it is required. + """ + + def __init__(self, value: str): + self._value = value + + def __repr__(self) -> str: + class_name = self.__class__.__name__ + return f"{class_name}('**********')" + + def __str__(self) -> str: + return self._value + + def __bool__(self) -> bool: + return bool(self._value) + + +class CommaSeparatedStrings(Sequence[str]): + def __init__(self, value: str | Sequence[str]): + if isinstance(value, str): + splitter = shlex(value, posix=True) + splitter.whitespace = "," + splitter.whitespace_split = True + self._items = [item.strip() for item in splitter] + else: + self._items = list(value) + + def __len__(self) -> int: + return len(self._items) + + def __getitem__(self, index: int | slice) -> Any: + return self._items[index] + + def __iter__(self) -> Iterator[str]: + return iter(self._items) + + def __repr__(self) -> str: + class_name = self.__class__.__name__ + items = [item for item in self] + return f"{class_name}({items!r})" + + def __str__(self) -> str: + return ", ".join(repr(item) for item in self) + + +class ImmutableMultiDict(Mapping[_KeyType, _CovariantValueType]): + _dict: dict[_KeyType, _CovariantValueType] + + def __init__( + self, + *args: ImmutableMultiDict[_KeyType, _CovariantValueType] + | Mapping[_KeyType, _CovariantValueType] + | Iterable[tuple[_KeyType, _CovariantValueType]], + **kwargs: Any, + ) -> None: + assert len(args) < 2, "Too many arguments." + + value: Any = args[0] if args else [] + if kwargs: + value = ImmutableMultiDict(value).multi_items() + ImmutableMultiDict(kwargs).multi_items() + + if not value: + _items: list[tuple[Any, Any]] = [] + elif hasattr(value, "multi_items"): + value = cast(ImmutableMultiDict[_KeyType, _CovariantValueType], value) + _items = list(value.multi_items()) + elif hasattr(value, "items"): + value = cast(Mapping[_KeyType, _CovariantValueType], value) + _items = list(value.items()) + else: + value = cast("list[tuple[Any, Any]]", value) + _items = list(value) + + self._dict = {k: v for k, v in _items} + self._list = _items + + def getlist(self, key: Any) -> list[_CovariantValueType]: + return [item_value for item_key, item_value in self._list if item_key == key] + + def keys(self) -> KeysView[_KeyType]: + return self._dict.keys() + + def values(self) -> ValuesView[_CovariantValueType]: + return self._dict.values() + + def items(self) -> ItemsView[_KeyType, _CovariantValueType]: + return self._dict.items() + + def multi_items(self) -> list[tuple[_KeyType, _CovariantValueType]]: + return list(self._list) + + def __getitem__(self, key: _KeyType) -> _CovariantValueType: + return self._dict[key] + + def __contains__(self, key: Any) -> bool: + return key in self._dict + + def __iter__(self) -> Iterator[_KeyType]: + return iter(self.keys()) + + def __len__(self) -> int: + return len(self._dict) + + def __eq__(self, other: Any) -> bool: + if not isinstance(other, self.__class__): + return False + return sorted(self._list) == sorted(other._list) + + def __repr__(self) -> str: + class_name = self.__class__.__name__ + items = self.multi_items() + return f"{class_name}({items!r})" + + +class MultiDict(ImmutableMultiDict[Any, Any]): + def __setitem__(self, key: Any, value: Any) -> None: + self.setlist(key, [value]) + + def __delitem__(self, key: Any) -> None: + self._list = [(k, v) for k, v in self._list if k != key] + del self._dict[key] + + def pop(self, key: Any, default: Any = None) -> Any: + self._list = [(k, v) for k, v in self._list if k != key] + return self._dict.pop(key, default) + + def popitem(self) -> tuple[Any, Any]: + key, value = self._dict.popitem() + self._list = [(k, v) for k, v in self._list if k != key] + return key, value + + def poplist(self, key: Any) -> list[Any]: + values = [v for k, v in self._list if k == key] + self.pop(key) + return values + + def clear(self) -> None: + self._dict.clear() + self._list.clear() + + def setdefault(self, key: Any, default: Any = None) -> Any: + if key not in self: + self._dict[key] = default + self._list.append((key, default)) + + return self[key] + + def setlist(self, key: Any, values: list[Any]) -> None: + if not values: + self.pop(key, None) + else: + existing_items = [(k, v) for (k, v) in self._list if k != key] + self._list = existing_items + [(key, value) for value in values] + self._dict[key] = values[-1] + + def append(self, key: Any, value: Any) -> None: + self._list.append((key, value)) + self._dict[key] = value + + def update( + self, + *args: MultiDict | Mapping[Any, Any] | list[tuple[Any, Any]], + **kwargs: Any, + ) -> None: + value = MultiDict(*args, **kwargs) + existing_items = [(k, v) for (k, v) in self._list if k not in value.keys()] + self._list = existing_items + value.multi_items() + self._dict.update(value) + + +class QueryParams(ImmutableMultiDict[str, str]): + """ + An immutable multidict. + """ + + def __init__( + self, + *args: ImmutableMultiDict[Any, Any] | Mapping[Any, Any] | list[tuple[Any, Any]] | str | bytes, + **kwargs: Any, + ) -> None: + assert len(args) < 2, "Too many arguments." + + value = args[0] if args else [] + + if isinstance(value, str): + super().__init__(parse_qsl(value, keep_blank_values=True), **kwargs) + elif isinstance(value, bytes): + super().__init__(parse_qsl(value.decode("latin-1"), keep_blank_values=True), **kwargs) + else: + super().__init__(*args, **kwargs) # type: ignore[arg-type] + self._list = [(str(k), str(v)) for k, v in self._list] + self._dict = {str(k): str(v) for k, v in self._dict.items()} + + def __str__(self) -> str: + return urlencode(self._list) + + def __repr__(self) -> str: + class_name = self.__class__.__name__ + query_string = str(self) + return f"{class_name}({query_string!r})" + + +class UploadFile: + """ + An uploaded file included as part of the request data. + """ + + def __init__( + self, + file: BinaryIO, + *, + size: int | None = None, + filename: str | None = None, + headers: Headers | None = None, + ) -> None: + self.filename = filename + self.file = file + self.size = size + self.headers = headers or Headers() + + # Capture max size from SpooledTemporaryFile if one is provided. This slightly speeds up future checks. + # Note 0 means unlimited mirroring SpooledTemporaryFile's __init__ + self._max_mem_size = getattr(self.file, "_max_size", 0) + + @property + def content_type(self) -> str | None: + return self.headers.get("content-type", None) + + @property + def _in_memory(self) -> bool: + # check for SpooledTemporaryFile._rolled + rolled_to_disk = getattr(self.file, "_rolled", True) + return not rolled_to_disk + + def _will_roll(self, size_to_add: int) -> bool: + # If we're not in_memory then we will always roll + if not self._in_memory: + return True + + # Check for SpooledTemporaryFile._max_size + future_size = self.file.tell() + size_to_add + return bool(future_size > self._max_mem_size) if self._max_mem_size else False + + async def write(self, data: bytes) -> None: + new_data_len = len(data) + if self.size is not None: + self.size += new_data_len + + if self._will_roll(new_data_len): + await run_in_threadpool(self.file.write, data) + else: + self.file.write(data) + + async def read(self, size: int = -1) -> bytes: + if self._in_memory: + return self.file.read(size) + return await run_in_threadpool(self.file.read, size) + + async def seek(self, offset: int) -> None: + if self._in_memory: + self.file.seek(offset) + else: + await run_in_threadpool(self.file.seek, offset) + + async def close(self) -> None: + if self._in_memory: + self.file.close() + else: + await run_in_threadpool(self.file.close) + + def __repr__(self) -> str: + return f"{self.__class__.__name__}(filename={self.filename!r}, size={self.size!r}, headers={self.headers!r})" + + +class FormData(ImmutableMultiDict[str, UploadFile | str]): + """ + An immutable multidict, containing both file uploads and text input. + """ + + def __init__( + self, + *args: FormData | Mapping[str, str | UploadFile] | list[tuple[str, str | UploadFile]], + **kwargs: str | UploadFile, + ) -> None: + super().__init__(*args, **kwargs) + + async def close(self) -> None: + for key, value in self.multi_items(): + if isinstance(value, UploadFile): + await value.close() + + +class Headers(Mapping[str, str]): + """ + An immutable, case-insensitive multidict. + """ + + def __init__( + self, + headers: Mapping[str, str] | None = None, + raw: list[tuple[bytes, bytes]] | None = None, + scope: MutableMapping[str, Any] | None = None, + ) -> None: + self._list: list[tuple[bytes, bytes]] = [] + if headers is not None: + assert raw is None, 'Cannot set both "headers" and "raw".' + assert scope is None, 'Cannot set both "headers" and "scope".' + self._list = [(key.lower().encode("latin-1"), value.encode("latin-1")) for key, value in headers.items()] + elif raw is not None: + assert scope is None, 'Cannot set both "raw" and "scope".' + self._list = raw + elif scope is not None: + # scope["headers"] isn't necessarily a list + # it might be a tuple or other iterable + self._list = scope["headers"] = list(scope["headers"]) + + @property + def raw(self) -> list[tuple[bytes, bytes]]: + return list(self._list) + + def keys(self) -> list[str]: # type: ignore[override] + return [key.decode("latin-1") for key, value in self._list] + + def values(self) -> list[str]: # type: ignore[override] + return [value.decode("latin-1") for key, value in self._list] + + def items(self) -> list[tuple[str, str]]: # type: ignore[override] + return [(key.decode("latin-1"), value.decode("latin-1")) for key, value in self._list] + + def getlist(self, key: str) -> list[str]: + get_header_key = key.lower().encode("latin-1") + return [item_value.decode("latin-1") for item_key, item_value in self._list if item_key == get_header_key] + + def mutablecopy(self) -> MutableHeaders: + return MutableHeaders(raw=self._list[:]) + + def __getitem__(self, key: str) -> str: + get_header_key = key.lower().encode("latin-1") + for header_key, header_value in self._list: + if header_key == get_header_key: + return header_value.decode("latin-1") + raise KeyError(key) + + def __contains__(self, key: Any) -> bool: + get_header_key = key.lower().encode("latin-1") + for header_key, header_value in self._list: + if header_key == get_header_key: + return True + return False + + def __iter__(self) -> Iterator[Any]: + return iter(self.keys()) + + def __len__(self) -> int: + return len(self._list) + + def __eq__(self, other: Any) -> bool: + if not isinstance(other, Headers): + return False + return sorted(self._list) == sorted(other._list) + + def __repr__(self) -> str: + class_name = self.__class__.__name__ + as_dict = dict(self.items()) + if len(as_dict) == len(self): + return f"{class_name}({as_dict!r})" + return f"{class_name}(raw={self.raw!r})" + + +class MutableHeaders(Headers): + def __setitem__(self, key: str, value: str) -> None: + """ + Set the header `key` to `value`, removing any duplicate entries. + Retains insertion order. + """ + set_key = key.lower().encode("latin-1") + set_value = value.encode("latin-1") + + found_indexes: list[int] = [] + for idx, (item_key, item_value) in enumerate(self._list): + if item_key == set_key: + found_indexes.append(idx) + + for idx in reversed(found_indexes[1:]): + del self._list[idx] + + if found_indexes: + idx = found_indexes[0] + self._list[idx] = (set_key, set_value) + else: + self._list.append((set_key, set_value)) + + def __delitem__(self, key: str) -> None: + """ + Remove the header `key`. + """ + del_key = key.lower().encode("latin-1") + + pop_indexes: list[int] = [] + for idx, (item_key, item_value) in enumerate(self._list): + if item_key == del_key: + pop_indexes.append(idx) + + for idx in reversed(pop_indexes): + del self._list[idx] + + def __ior__(self, other: Mapping[str, str]) -> MutableHeaders: + if not isinstance(other, Mapping): + raise TypeError(f"Expected a mapping but got {other.__class__.__name__}") + self.update(other) + return self + + def __or__(self, other: Mapping[str, str]) -> MutableHeaders: + if not isinstance(other, Mapping): + raise TypeError(f"Expected a mapping but got {other.__class__.__name__}") + new = self.mutablecopy() + new.update(other) + return new + + @property + def raw(self) -> list[tuple[bytes, bytes]]: + return self._list + + def setdefault(self, key: str, value: str) -> str: + """ + If the header `key` does not exist, then set it to `value`. + Returns the header value. + """ + set_key = key.lower().encode("latin-1") + set_value = value.encode("latin-1") + + for idx, (item_key, item_value) in enumerate(self._list): + if item_key == set_key: + return item_value.decode("latin-1") + self._list.append((set_key, set_value)) + return value + + def update(self, other: Mapping[str, str]) -> None: + for key, val in other.items(): + self[key] = val + + def append(self, key: str, value: str) -> None: + """ + Append a header, preserving any duplicate entries. + """ + append_key = key.lower().encode("latin-1") + append_value = value.encode("latin-1") + self._list.append((append_key, append_value)) + + def add_vary_header(self, vary: str) -> None: + existing = self.get("vary") + if existing is not None: + vary = ", ".join([existing, vary]) + self["vary"] = vary + + +class State: + """ + An object that can be used to store arbitrary state. + + Used for `request.state` and `app.state`. + """ + + _state: dict[str, Any] + + def __init__(self, state: dict[str, Any] | None = None): + if state is None: + state = {} + super().__setattr__("_state", state) + + def __setattr__(self, key: Any, value: Any) -> None: + self._state[key] = value + + def __getattr__(self, key: Any) -> Any: + try: + return self._state[key] + except KeyError: + message = "'{}' object has no attribute '{}'" + raise AttributeError(message.format(self.__class__.__name__, key)) + + def __delattr__(self, key: Any) -> None: + del self._state[key] + + def __getitem__(self, key: str) -> Any: + return self._state[key] + + def __setitem__(self, key: str, value: Any) -> None: + self._state[key] = value + + def __delitem__(self, key: str) -> None: + del self._state[key] + + def __iter__(self) -> Iterator[str]: + return iter(self._state) + + def __len__(self) -> int: + return len(self._state)