base.py 7.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242
  1. import email.message
  2. import json
  3. import logging
  4. import re
  5. from typing import (
  6. TYPE_CHECKING,
  7. Collection,
  8. Container,
  9. Iterable,
  10. Iterator,
  11. List,
  12. Optional,
  13. Union,
  14. )
  15. from pip._vendor.packaging.requirements import Requirement
  16. from pip._vendor.packaging.version import LegacyVersion, Version
  17. from pip._internal.models.direct_url import (
  18. DIRECT_URL_METADATA_NAME,
  19. DirectUrl,
  20. DirectUrlValidationError,
  21. )
  22. from pip._internal.utils.misc import stdlib_pkgs # TODO: Move definition here.
  23. if TYPE_CHECKING:
  24. from typing import Protocol
  25. from pip._vendor.packaging.utils import NormalizedName
  26. else:
  27. Protocol = object
  28. DistributionVersion = Union[LegacyVersion, Version]
  29. logger = logging.getLogger(__name__)
  30. class BaseEntryPoint(Protocol):
  31. @property
  32. def name(self) -> str:
  33. raise NotImplementedError()
  34. @property
  35. def value(self) -> str:
  36. raise NotImplementedError()
  37. @property
  38. def group(self) -> str:
  39. raise NotImplementedError()
  40. class BaseDistribution(Protocol):
  41. @property
  42. def location(self) -> Optional[str]:
  43. """Where the distribution is loaded from.
  44. A string value is not necessarily a filesystem path, since distributions
  45. can be loaded from other sources, e.g. arbitrary zip archives. ``None``
  46. means the distribution is created in-memory.
  47. Do not canonicalize this value with e.g. ``pathlib.Path.resolve()``. If
  48. this is a symbolic link, we want to preserve the relative path between
  49. it and files in the distribution.
  50. """
  51. raise NotImplementedError()
  52. @property
  53. def info_directory(self) -> Optional[str]:
  54. """Location of the .[egg|dist]-info directory.
  55. Similarly to ``location``, a string value is not necessarily a
  56. filesystem path. ``None`` means the distribution is created in-memory.
  57. For a modern .dist-info installation on disk, this should be something
  58. like ``{location}/{raw_name}-{version}.dist-info``.
  59. Do not canonicalize this value with e.g. ``pathlib.Path.resolve()``. If
  60. this is a symbolic link, we want to preserve the relative path between
  61. it and other files in the distribution.
  62. """
  63. raise NotImplementedError()
  64. @property
  65. def canonical_name(self) -> "NormalizedName":
  66. raise NotImplementedError()
  67. @property
  68. def version(self) -> DistributionVersion:
  69. raise NotImplementedError()
  70. @property
  71. def direct_url(self) -> Optional[DirectUrl]:
  72. """Obtain a DirectUrl from this distribution.
  73. Returns None if the distribution has no `direct_url.json` metadata,
  74. or if `direct_url.json` is invalid.
  75. """
  76. try:
  77. content = self.read_text(DIRECT_URL_METADATA_NAME)
  78. except FileNotFoundError:
  79. return None
  80. try:
  81. return DirectUrl.from_json(content)
  82. except (
  83. UnicodeDecodeError,
  84. json.JSONDecodeError,
  85. DirectUrlValidationError,
  86. ) as e:
  87. logger.warning(
  88. "Error parsing %s for %s: %s",
  89. DIRECT_URL_METADATA_NAME,
  90. self.canonical_name,
  91. e,
  92. )
  93. return None
  94. @property
  95. def installer(self) -> str:
  96. raise NotImplementedError()
  97. @property
  98. def editable(self) -> bool:
  99. raise NotImplementedError()
  100. @property
  101. def local(self) -> bool:
  102. raise NotImplementedError()
  103. @property
  104. def in_usersite(self) -> bool:
  105. raise NotImplementedError()
  106. @property
  107. def in_site_packages(self) -> bool:
  108. raise NotImplementedError()
  109. def read_text(self, name: str) -> str:
  110. """Read a file in the .dist-info (or .egg-info) directory.
  111. Should raise ``FileNotFoundError`` if ``name`` does not exist in the
  112. metadata directory.
  113. """
  114. raise NotImplementedError()
  115. def iter_entry_points(self) -> Iterable[BaseEntryPoint]:
  116. raise NotImplementedError()
  117. @property
  118. def metadata(self) -> email.message.Message:
  119. """Metadata of distribution parsed from e.g. METADATA or PKG-INFO."""
  120. raise NotImplementedError()
  121. @property
  122. def metadata_version(self) -> Optional[str]:
  123. """Value of "Metadata-Version:" in distribution metadata, if available."""
  124. return self.metadata.get("Metadata-Version")
  125. @property
  126. def raw_name(self) -> str:
  127. """Value of "Name:" in distribution metadata."""
  128. # The metadata should NEVER be missing the Name: key, but if it somehow
  129. # does not, fall back to the known canonical name.
  130. return self.metadata.get("Name", self.canonical_name)
  131. def iter_dependencies(self, extras: Collection[str] = ()) -> Iterable[Requirement]:
  132. raise NotImplementedError()
  133. class BaseEnvironment:
  134. """An environment containing distributions to introspect."""
  135. @classmethod
  136. def default(cls) -> "BaseEnvironment":
  137. raise NotImplementedError()
  138. @classmethod
  139. def from_paths(cls, paths: Optional[List[str]]) -> "BaseEnvironment":
  140. raise NotImplementedError()
  141. def get_distribution(self, name: str) -> Optional["BaseDistribution"]:
  142. """Given a requirement name, return the installed distributions."""
  143. raise NotImplementedError()
  144. def _iter_distributions(self) -> Iterator["BaseDistribution"]:
  145. """Iterate through installed distributions.
  146. This function should be implemented by subclass, but never called
  147. directly. Use the public ``iter_distribution()`` instead, which
  148. implements additional logic to make sure the distributions are valid.
  149. """
  150. raise NotImplementedError()
  151. def iter_distributions(self) -> Iterator["BaseDistribution"]:
  152. """Iterate through installed distributions."""
  153. for dist in self._iter_distributions():
  154. # Make sure the distribution actually comes from a valid Python
  155. # packaging distribution. Pip's AdjacentTempDirectory leaves folders
  156. # e.g. ``~atplotlib.dist-info`` if cleanup was interrupted. The
  157. # valid project name pattern is taken from PEP 508.
  158. project_name_valid = re.match(
  159. r"^([A-Z0-9]|[A-Z0-9][A-Z0-9._-]*[A-Z0-9])$",
  160. dist.canonical_name,
  161. flags=re.IGNORECASE,
  162. )
  163. if not project_name_valid:
  164. logger.warning(
  165. "Ignoring invalid distribution %s (%s)",
  166. dist.canonical_name,
  167. dist.location,
  168. )
  169. continue
  170. yield dist
  171. def iter_installed_distributions(
  172. self,
  173. local_only: bool = True,
  174. skip: Container[str] = stdlib_pkgs,
  175. include_editables: bool = True,
  176. editables_only: bool = False,
  177. user_only: bool = False,
  178. ) -> Iterator[BaseDistribution]:
  179. """Return a list of installed distributions.
  180. :param local_only: If True (default), only return installations
  181. local to the current virtualenv, if in a virtualenv.
  182. :param skip: An iterable of canonicalized project names to ignore;
  183. defaults to ``stdlib_pkgs``.
  184. :param include_editables: If False, don't report editables.
  185. :param editables_only: If True, only report editables.
  186. :param user_only: If True, only report installations in the user
  187. site directory.
  188. """
  189. it = self.iter_distributions()
  190. if local_only:
  191. it = (d for d in it if d.local)
  192. if not include_editables:
  193. it = (d for d in it if not d.editable)
  194. if editables_only:
  195. it = (d for d in it if d.editable)
  196. if user_only:
  197. it = (d for d in it if d.in_usersite)
  198. return (d for d in it if d.canonical_name not in skip)