From 6d7ba58f880be618ade07f8ea080fe8c4bf8a896 Mon Sep 17 00:00:00 2001 From: cyfraeviolae Date: Wed, 3 Apr 2024 03:10:44 -0400 Subject: venv --- .../site-packages/litestar/datastructures/url.py | 262 +++++++++++++++++++++ 1 file changed, 262 insertions(+) create mode 100644 venv/lib/python3.11/site-packages/litestar/datastructures/url.py (limited to 'venv/lib/python3.11/site-packages/litestar/datastructures/url.py') diff --git a/venv/lib/python3.11/site-packages/litestar/datastructures/url.py b/venv/lib/python3.11/site-packages/litestar/datastructures/url.py new file mode 100644 index 0000000..f3441d0 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/datastructures/url.py @@ -0,0 +1,262 @@ +from __future__ import annotations + +from functools import lru_cache +from typing import TYPE_CHECKING, Any, NamedTuple +from urllib.parse import SplitResult, urlencode, urlsplit, urlunsplit + +from litestar._parsers import parse_query_string +from litestar.datastructures import MultiDict +from litestar.types import Empty + +if TYPE_CHECKING: + from typing_extensions import Self + + from litestar.types import EmptyType, Scope + +__all__ = ("Address", "URL") + +_DEFAULT_SCHEME_PORTS = {"http": 80, "https": 443, "ftp": 21, "ws": 80, "wss": 443} + + +class Address(NamedTuple): + """Just a network address.""" + + host: str + """Address host.""" + port: int + """Address port.""" + + +def make_absolute_url(path: str | URL, base: str | URL) -> str: + """Create an absolute URL. + + Args: + path: URL path to make absolute + base: URL to use as a base + + Returns: + A string representing the new, absolute URL + """ + url = base if isinstance(base, URL) else URL(base) + netloc = url.netloc + path = url.path.rstrip("/") + str(path) + return str(URL.from_components(scheme=url.scheme, netloc=netloc, path=path)) + + +class URL: + """Representation and modification utilities of a URL.""" + + __slots__ = ( + "_query_params", + "_parsed_url", + "fragment", + "hostname", + "netloc", + "password", + "path", + "port", + "query", + "scheme", + "username", + ) + + _query_params: EmptyType | MultiDict + _parsed_url: str | None + + scheme: str + """URL scheme.""" + netloc: str + """Network location.""" + path: str + """Hierarchical path.""" + fragment: str + """Fragment component.""" + query: str + """Query string.""" + username: str | None + """Username if specified.""" + password: str | None + """Password if specified.""" + port: int | None + """Port if specified.""" + hostname: str | None + """Hostname if specified.""" + + def __new__(cls, url: str | SplitResult) -> URL: + """Create a new instance. + + Args: + url: url string or split result to represent. + """ + return cls._new(url=url) + + @classmethod + @lru_cache + def _new(cls, url: str | SplitResult) -> URL: + instance = super().__new__(cls) + instance._parsed_url = None + + if isinstance(url, str): + result = urlsplit(url) + instance._parsed_url = url + else: + result = url + + instance.scheme = result.scheme + instance.netloc = result.netloc + instance.path = result.path + instance.fragment = result.fragment + instance.query = result.query + instance.username = result.username + instance.password = result.password + instance.port = result.port + instance.hostname = result.hostname + instance._query_params = Empty + + return instance + + @property + def _url(self) -> str: + if not self._parsed_url: + self._parsed_url = str( + urlunsplit( + SplitResult( + scheme=self.scheme, + netloc=self.netloc, + path=self.path, + fragment=self.fragment, + query=self.query, + ) + ) + ) + return self._parsed_url + + @classmethod + @lru_cache + def from_components( + cls, + scheme: str = "", + netloc: str = "", + path: str = "", + fragment: str = "", + query: str = "", + ) -> Self: + """Create a new URL from components. + + Args: + scheme: URL scheme + netloc: Network location + path: Hierarchical path + query: Query component + fragment: Fragment identifier + + Returns: + A new URL with the given components + """ + return cls( + SplitResult( + scheme=scheme, + netloc=netloc, + path=path, + fragment=fragment, + query=query, + ) + ) + + @classmethod + def from_scope(cls, scope: Scope) -> Self: + """Construct a URL from a :class:`Scope <.types.Scope>` + + Args: + scope: A scope + + Returns: + A URL + """ + scheme = scope.get("scheme", "http") + server = scope.get("server") + path = scope.get("root_path", "") + scope["path"] + query_string = scope.get("query_string", b"") + + # we use iteration here because it's faster, and headers might not yet be cached + host = next( + ( + header_value.decode("latin-1") + for header_name, header_value in scope.get("headers", []) + if header_name == b"host" + ), + "", + ) + if server and not host: + host, port = server + default_port = _DEFAULT_SCHEME_PORTS[scheme] + if port != default_port: + host = f"{host}:{port}" + + return cls.from_components( + scheme=scheme if server else "", + query=query_string.decode(), + netloc=host, + path=path, + ) + + def with_replacements( + self, + scheme: str = "", + netloc: str = "", + path: str = "", + query: str | MultiDict | None | EmptyType = Empty, + fragment: str = "", + ) -> Self: + """Create a new URL, replacing the given components. + + Args: + scheme: URL scheme + netloc: Network location + path: Hierarchical path + query: Raw query string + fragment: Fragment identifier + + Returns: + A new URL with the given components replaced + """ + if isinstance(query, MultiDict): + query = urlencode(query=query) + + query = (query if query is not Empty else self.query) or "" + + return type(self).from_components( + scheme=scheme or self.scheme, + netloc=netloc or self.netloc, + path=path or self.path, + query=query, + fragment=fragment or self.fragment, + ) + + @property + def query_params(self) -> MultiDict: + """Query parameters of a URL as a :class:`MultiDict <.datastructures.multi_dicts.MultiDict>` + + Returns: + A :class:`MultiDict <.datastructures.multi_dicts.MultiDict>` with query parameters + + Notes: + - The returned ``MultiDict`` is mutable, :class:`URL` itself is *immutable*, + therefore mutating the query parameters will not directly mutate the ``URL``. + If you want to modify query parameters, make modifications in the + multidict and pass them back to :meth:`with_replacements` + """ + if self._query_params is Empty: + self._query_params = MultiDict(parse_query_string(query_string=self.query.encode())) + return self._query_params + + def __str__(self) -> str: + return self._url + + def __eq__(self, other: Any) -> bool: + if isinstance(other, (str, URL)): + return str(self) == str(other) + return NotImplemented # pragma: no cover + + def __repr__(self) -> str: + return f"{type(self).__name__}({self._url!r})" -- cgit v1.2.3