1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
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})"
|
