direct_url.py 6.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224
  1. """ PEP 610 """
  2. import json
  3. import re
  4. import urllib.parse
  5. from dataclasses import dataclass
  6. from typing import Any, ClassVar, Dict, Iterable, Optional, Type, TypeVar, Union
  7. __all__ = [
  8. "DirectUrl",
  9. "DirectUrlValidationError",
  10. "DirInfo",
  11. "ArchiveInfo",
  12. "VcsInfo",
  13. ]
  14. T = TypeVar("T")
  15. DIRECT_URL_METADATA_NAME = "direct_url.json"
  16. ENV_VAR_RE = re.compile(r"^\$\{[A-Za-z0-9-_]+\}(:\$\{[A-Za-z0-9-_]+\})?$")
  17. class DirectUrlValidationError(Exception):
  18. pass
  19. def _get(
  20. d: Dict[str, Any], expected_type: Type[T], key: str, default: Optional[T] = None
  21. ) -> Optional[T]:
  22. """Get value from dictionary and verify expected type."""
  23. if key not in d:
  24. return default
  25. value = d[key]
  26. if not isinstance(value, expected_type):
  27. raise DirectUrlValidationError(
  28. f"{value!r} has unexpected type for {key} (expected {expected_type})"
  29. )
  30. return value
  31. def _get_required(
  32. d: Dict[str, Any], expected_type: Type[T], key: str, default: Optional[T] = None
  33. ) -> T:
  34. value = _get(d, expected_type, key, default)
  35. if value is None:
  36. raise DirectUrlValidationError(f"{key} must have a value")
  37. return value
  38. def _exactly_one_of(infos: Iterable[Optional["InfoType"]]) -> "InfoType":
  39. infos = [info for info in infos if info is not None]
  40. if not infos:
  41. raise DirectUrlValidationError(
  42. "missing one of archive_info, dir_info, vcs_info"
  43. )
  44. if len(infos) > 1:
  45. raise DirectUrlValidationError(
  46. "more than one of archive_info, dir_info, vcs_info"
  47. )
  48. assert infos[0] is not None
  49. return infos[0]
  50. def _filter_none(**kwargs: Any) -> Dict[str, Any]:
  51. """Make dict excluding None values."""
  52. return {k: v for k, v in kwargs.items() if v is not None}
  53. @dataclass
  54. class VcsInfo:
  55. name: ClassVar = "vcs_info"
  56. vcs: str
  57. commit_id: str
  58. requested_revision: Optional[str] = None
  59. @classmethod
  60. def _from_dict(cls, d: Optional[Dict[str, Any]]) -> Optional["VcsInfo"]:
  61. if d is None:
  62. return None
  63. return cls(
  64. vcs=_get_required(d, str, "vcs"),
  65. commit_id=_get_required(d, str, "commit_id"),
  66. requested_revision=_get(d, str, "requested_revision"),
  67. )
  68. def _to_dict(self) -> Dict[str, Any]:
  69. return _filter_none(
  70. vcs=self.vcs,
  71. requested_revision=self.requested_revision,
  72. commit_id=self.commit_id,
  73. )
  74. class ArchiveInfo:
  75. name = "archive_info"
  76. def __init__(
  77. self,
  78. hash: Optional[str] = None,
  79. hashes: Optional[Dict[str, str]] = None,
  80. ) -> None:
  81. # set hashes before hash, since the hash setter will further populate hashes
  82. self.hashes = hashes
  83. self.hash = hash
  84. @property
  85. def hash(self) -> Optional[str]:
  86. return self._hash
  87. @hash.setter
  88. def hash(self, value: Optional[str]) -> None:
  89. if value is not None:
  90. # Auto-populate the hashes key to upgrade to the new format automatically.
  91. # We don't back-populate the legacy hash key from hashes.
  92. try:
  93. hash_name, hash_value = value.split("=", 1)
  94. except ValueError:
  95. raise DirectUrlValidationError(
  96. f"invalid archive_info.hash format: {value!r}"
  97. )
  98. if self.hashes is None:
  99. self.hashes = {hash_name: hash_value}
  100. elif hash_name not in self.hashes:
  101. self.hashes = self.hashes.copy()
  102. self.hashes[hash_name] = hash_value
  103. self._hash = value
  104. @classmethod
  105. def _from_dict(cls, d: Optional[Dict[str, Any]]) -> Optional["ArchiveInfo"]:
  106. if d is None:
  107. return None
  108. return cls(hash=_get(d, str, "hash"), hashes=_get(d, dict, "hashes"))
  109. def _to_dict(self) -> Dict[str, Any]:
  110. return _filter_none(hash=self.hash, hashes=self.hashes)
  111. @dataclass
  112. class DirInfo:
  113. name: ClassVar = "dir_info"
  114. editable: bool = False
  115. @classmethod
  116. def _from_dict(cls, d: Optional[Dict[str, Any]]) -> Optional["DirInfo"]:
  117. if d is None:
  118. return None
  119. return cls(editable=_get_required(d, bool, "editable", default=False))
  120. def _to_dict(self) -> Dict[str, Any]:
  121. return _filter_none(editable=self.editable or None)
  122. InfoType = Union[ArchiveInfo, DirInfo, VcsInfo]
  123. @dataclass
  124. class DirectUrl:
  125. url: str
  126. info: InfoType
  127. subdirectory: Optional[str] = None
  128. def _remove_auth_from_netloc(self, netloc: str) -> str:
  129. if "@" not in netloc:
  130. return netloc
  131. user_pass, netloc_no_user_pass = netloc.split("@", 1)
  132. if (
  133. isinstance(self.info, VcsInfo)
  134. and self.info.vcs == "git"
  135. and user_pass == "git"
  136. ):
  137. return netloc
  138. if ENV_VAR_RE.match(user_pass):
  139. return netloc
  140. return netloc_no_user_pass
  141. @property
  142. def redacted_url(self) -> str:
  143. """url with user:password part removed unless it is formed with
  144. environment variables as specified in PEP 610, or it is ``git``
  145. in the case of a git URL.
  146. """
  147. purl = urllib.parse.urlsplit(self.url)
  148. netloc = self._remove_auth_from_netloc(purl.netloc)
  149. surl = urllib.parse.urlunsplit(
  150. (purl.scheme, netloc, purl.path, purl.query, purl.fragment)
  151. )
  152. return surl
  153. def validate(self) -> None:
  154. self.from_dict(self.to_dict())
  155. @classmethod
  156. def from_dict(cls, d: Dict[str, Any]) -> "DirectUrl":
  157. return DirectUrl(
  158. url=_get_required(d, str, "url"),
  159. subdirectory=_get(d, str, "subdirectory"),
  160. info=_exactly_one_of(
  161. [
  162. ArchiveInfo._from_dict(_get(d, dict, "archive_info")),
  163. DirInfo._from_dict(_get(d, dict, "dir_info")),
  164. VcsInfo._from_dict(_get(d, dict, "vcs_info")),
  165. ]
  166. ),
  167. )
  168. def to_dict(self) -> Dict[str, Any]:
  169. res = _filter_none(
  170. url=self.redacted_url,
  171. subdirectory=self.subdirectory,
  172. )
  173. res[self.info.name] = self.info._to_dict()
  174. return res
  175. @classmethod
  176. def from_json(cls, s: str) -> "DirectUrl":
  177. return cls.from_dict(json.loads(s))
  178. def to_json(self) -> str:
  179. return json.dumps(self.to_dict(), sort_keys=True)
  180. def is_local_editable(self) -> bool:
  181. return isinstance(self.info, DirInfo) and self.info.editable