Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import dataclasses
4import re
5import urllib.parse
6from collections.abc import Mapping
7from typing import TYPE_CHECKING, Any, Protocol, TypeVar
9if TYPE_CHECKING: # pragma: no cover
10 import sys
11 from collections.abc import Collection
12 from urllib.parse import SplitResult
14 if sys.version_info >= (3, 11):
15 from typing import Self
16 else:
17 from typing_extensions import Self
19__all__ = [
20 "ArchiveInfo",
21 "DirInfo",
22 "DirectUrl",
23 "DirectUrlValidationError",
24 "VcsInfo",
25]
28def __dir__() -> list[str]:
29 return __all__
32_T = TypeVar("_T")
35class _FromMappingProtocol(Protocol): # pragma: no cover
36 @classmethod
37 def _from_dict(cls, d: Mapping[str, Any]) -> Self: ...
40_FromMappingProtocolT = TypeVar("_FromMappingProtocolT", bound=_FromMappingProtocol)
43def _json_dict_factory(data: list[tuple[str, Any]]) -> dict[str, Any]:
44 return {key: value for key, value in data if value is not None}
47def _get(d: Mapping[str, Any], expected_type: type[_T], key: str) -> _T | None:
48 """Get a value from the dictionary and verify it's the expected type."""
49 if (value := d.get(key)) is None:
50 return None
51 if not isinstance(value, expected_type):
52 raise DirectUrlValidationError(
53 f"Unexpected type {type(value).__name__} "
54 f"(expected {expected_type.__name__})",
55 context=key,
56 )
57 return value
60def _get_required(d: Mapping[str, Any], expected_type: type[_T], key: str) -> _T:
61 """Get a required value from the dictionary and verify it's the expected type."""
62 if (value := _get(d, expected_type, key)) is None:
63 raise _DirectUrlRequiredKeyError(key)
64 return value
67def _get_object(
68 d: Mapping[str, Any], target_type: type[_FromMappingProtocolT], key: str
69) -> _FromMappingProtocolT | None:
70 """Get a dictionary value from the dictionary and convert it to a dataclass."""
71 if (value := _get(d, Mapping, key)) is None: # type: ignore[type-abstract]
72 return None
73 try:
74 return target_type._from_dict(value)
75 except Exception as e:
76 raise DirectUrlValidationError(e, context=key) from e
79_PEP610_USER_PASS_ENV_VARS_REGEX = re.compile(
80 r"^\$\{[A-Za-z0-9-_]+\}(:\$\{[A-Za-z0-9-_]+\})?$"
81)
84def _strip_auth_from_netloc(netloc: str, safe_user_passwords: Collection[str]) -> str:
85 if "@" not in netloc:
86 return netloc
87 user_pass, netloc_no_user_pass = netloc.rsplit("@", 1)
88 if user_pass in safe_user_passwords:
89 return netloc
90 if _PEP610_USER_PASS_ENV_VARS_REGEX.match(user_pass):
91 return netloc
92 return netloc_no_user_pass
95def _strip_url(url: str, safe_user_passwords: Collection[str]) -> str:
96 """url with user:password part removed unless it is formed with
97 environment variables as specified in PEP 610, or it is a safe user:password
98 such as `git`.
99 """
100 parsed_url = urllib.parse.urlsplit(url)
101 netloc = _strip_auth_from_netloc(parsed_url.netloc, safe_user_passwords)
102 return urllib.parse.urlunsplit(
103 (
104 parsed_url.scheme,
105 netloc,
106 parsed_url.path,
107 parsed_url.query,
108 parsed_url.fragment,
109 )
110 )
113def _file_url_has_absolute_path(parsed_url: SplitResult) -> bool:
114 return parsed_url.path.startswith("/")
117class DirectUrlValidationError(Exception):
118 """Raised when when input data is not spec-compliant.
120 .. versionadded:: 26.1
121 """
123 context: str | None = None
124 message: str
126 def __init__(
127 self,
128 cause: str | Exception,
129 *,
130 context: str | None = None,
131 ) -> None:
132 if isinstance(cause, DirectUrlValidationError):
133 if cause.context:
134 self.context = (
135 f"{context}.{cause.context}" if context else cause.context
136 )
137 else:
138 self.context = context # pragma: no cover
139 self.message = cause.message
140 else:
141 self.context = context
142 self.message = str(cause)
144 def __str__(self) -> str:
145 if self.context:
146 return f"{self.message} in {self.context!r}"
147 return self.message
150class _DirectUrlRequiredKeyError(DirectUrlValidationError):
151 def __init__(self, key: str) -> None:
152 super().__init__("Missing required value", context=key)
155@dataclasses.dataclass(frozen=True, init=False)
156class VcsInfo:
157 """The version control information of a :class:`DirectUrl`."""
159 vcs: str
160 commit_id: str
161 requested_revision: str | None = None
163 def __init__(
164 self,
165 *,
166 vcs: str,
167 commit_id: str,
168 requested_revision: str | None = None,
169 ) -> None:
170 object.__setattr__(self, "vcs", vcs)
171 object.__setattr__(self, "commit_id", commit_id)
172 object.__setattr__(self, "requested_revision", requested_revision)
174 @classmethod
175 def _from_dict(cls, d: Mapping[str, Any]) -> Self:
176 # We can't validate vcs value because is not closed.
177 return cls(
178 vcs=_get_required(d, str, "vcs"),
179 requested_revision=_get(d, str, "requested_revision"),
180 commit_id=_get_required(d, str, "commit_id"),
181 )
184@dataclasses.dataclass(frozen=True, init=False)
185class ArchiveInfo:
186 """The archive information of a :class:`DirectUrl`."""
188 hashes: Mapping[str, str] | None = None
190 def __init__(
191 self,
192 *,
193 hashes: Mapping[str, str] | None = None,
194 ) -> None:
195 object.__setattr__(self, "hashes", hashes)
197 @classmethod
198 def _from_dict(cls, d: Mapping[str, Any]) -> Self:
199 hashes = _get(d, Mapping, "hashes") # type: ignore[type-abstract]
200 if hashes is not None and not all(isinstance(h, str) for h in hashes.values()):
201 raise DirectUrlValidationError(
202 "Hash values must be strings", context="hashes"
203 )
204 legacy_hash = _get(d, str, "hash")
205 if legacy_hash is not None:
206 if "=" not in legacy_hash:
207 raise DirectUrlValidationError(
208 "Invalid hash format (expected '<algorithm>=<hash>')",
209 context="hash",
210 )
211 hash_algorithm, hash_value = legacy_hash.split("=", 1)
212 if hashes is None:
213 # if `hashes` are not present, we can derive it from the legacy `hash`
214 hashes = {hash_algorithm: hash_value}
215 else:
216 # if `hashes` are present, the legacy `hash` must match one of them
217 if hash_algorithm not in hashes:
218 raise DirectUrlValidationError(
219 f"Algorithm {hash_algorithm!r} used in hash field "
220 f"is not present in hashes field",
221 context="hashes",
222 )
223 if hashes[hash_algorithm] != hash_value:
224 raise DirectUrlValidationError(
225 f"Algorithm {hash_algorithm!r} used in hash field "
226 f"has different value in hashes field",
227 context="hash",
228 )
229 return cls(hashes=hashes)
232@dataclasses.dataclass(frozen=True, init=False)
233class DirInfo:
234 """The local directory information of a :class:`DirectUrl`."""
236 editable: bool | None = None
238 def __init__(
239 self,
240 *,
241 editable: bool | None = None,
242 ) -> None:
243 object.__setattr__(self, "editable", editable)
245 @classmethod
246 def _from_dict(cls, d: Mapping[str, Any]) -> Self:
247 return cls(
248 editable=_get(d, bool, "editable"),
249 )
252@dataclasses.dataclass(frozen=True, init=False)
253class DirectUrl:
254 """A class representing a direct URL.
256 .. versionadded:: 26.1
257 """
259 url: str
260 archive_info: ArchiveInfo | None = None
261 vcs_info: VcsInfo | None = None
262 dir_info: DirInfo | None = None
263 subdirectory: str | None = None # XXX Path or str?
265 def __init__(
266 self,
267 *,
268 url: str,
269 archive_info: ArchiveInfo | None = None,
270 vcs_info: VcsInfo | None = None,
271 dir_info: DirInfo | None = None,
272 subdirectory: str | None = None,
273 ) -> None:
274 object.__setattr__(self, "url", url)
275 object.__setattr__(self, "archive_info", archive_info)
276 object.__setattr__(self, "vcs_info", vcs_info)
277 object.__setattr__(self, "dir_info", dir_info)
278 object.__setattr__(self, "subdirectory", subdirectory)
280 @classmethod
281 def _from_dict(cls, d: Mapping[str, Any]) -> Self:
282 direct_url = cls(
283 url=_get_required(d, str, "url"),
284 archive_info=_get_object(d, ArchiveInfo, "archive_info"),
285 vcs_info=_get_object(d, VcsInfo, "vcs_info"),
286 dir_info=_get_object(d, DirInfo, "dir_info"),
287 subdirectory=_get(d, str, "subdirectory"),
288 )
289 if (
290 bool(direct_url.vcs_info)
291 + bool(direct_url.archive_info)
292 + bool(direct_url.dir_info)
293 ) != 1:
294 raise DirectUrlValidationError(
295 "Exactly one of vcs_info, archive_info, dir_info must be present"
296 )
297 if direct_url.dir_info is not None:
298 parsed_url = urllib.parse.urlsplit(direct_url.url)
299 if parsed_url.scheme != "file":
300 raise DirectUrlValidationError(
301 "URL scheme must be file:// when dir_info is present",
302 context="url",
303 )
304 if not _file_url_has_absolute_path(parsed_url):
305 raise DirectUrlValidationError(
306 "File URL must be absolute when dir_info is present",
307 context="url",
308 )
309 # XXX subdirectory must be relative, can we, should we validate that here?
310 return direct_url
312 @classmethod
313 def from_dict(cls, d: Mapping[str, Any], /) -> Self:
314 """Create and validate a DirectUrl instance from a JSON dictionary."""
315 return cls._from_dict(d)
317 def to_dict(
318 self,
319 *,
320 generate_legacy_hash: bool = False,
321 strip_user_password: bool = True,
322 safe_user_passwords: Collection[str] = ("git",),
323 ) -> Mapping[str, Any]:
324 """Convert the DirectUrl instance to a JSON dictionary.
326 :param generate_legacy_hash: If True, include a legacy `hash` field in
327 `archive_info` for backward compatibility with tools that don't
328 support the `hashes` field.
329 :param strip_user_password: If True, strip user:password from the URL
330 unless it is formed with environment variables as specified in PEP
331 610, or it is a safe user:password such as `git`.
332 :param safe_user_passwords: A collection of user:password strings that
333 should not be stripped from the URL even if `strip_user_password` is
334 True.
335 """
336 res = dataclasses.asdict(self, dict_factory=_json_dict_factory)
337 if generate_legacy_hash and self.archive_info and self.archive_info.hashes:
338 hash_algorithm, hash_value = next(iter(self.archive_info.hashes.items()))
339 res["archive_info"]["hash"] = f"{hash_algorithm}={hash_value}"
340 if strip_user_password:
341 res["url"] = _strip_url(self.url, safe_user_passwords)
342 return res
344 def validate(self) -> None:
345 """Validate the DirectUrl instance against the specification.
347 Raises :class:`DirectUrlValidationError` if invalid.
348 """
349 self.from_dict(self.to_dict())