pkg_resources.py 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298
  1. from __future__ import annotations
  2. import email.message
  3. import email.parser
  4. import logging
  5. import os
  6. import zipfile
  7. from collections.abc import Collection, Iterable, Iterator, Mapping
  8. from typing import (
  9. NamedTuple,
  10. )
  11. from pip._vendor import pkg_resources
  12. from pip._vendor.packaging.requirements import Requirement
  13. from pip._vendor.packaging.utils import NormalizedName, canonicalize_name
  14. from pip._vendor.packaging.version import Version
  15. from pip._vendor.packaging.version import parse as parse_version
  16. from pip._internal.exceptions import InvalidWheel, NoneMetadataError, UnsupportedWheel
  17. from pip._internal.utils.egg_link import egg_link_path_from_location
  18. from pip._internal.utils.misc import display_path, normalize_path
  19. from pip._internal.utils.wheel import parse_wheel, read_wheel_metadata_file
  20. from .base import (
  21. BaseDistribution,
  22. BaseEntryPoint,
  23. BaseEnvironment,
  24. InfoPath,
  25. Wheel,
  26. )
  27. __all__ = ["NAME", "Distribution", "Environment"]
  28. logger = logging.getLogger(__name__)
  29. NAME = "pkg_resources"
  30. class EntryPoint(NamedTuple):
  31. name: str
  32. value: str
  33. group: str
  34. class InMemoryMetadata:
  35. """IMetadataProvider that reads metadata files from a dictionary.
  36. This also maps metadata decoding exceptions to our internal exception type.
  37. """
  38. def __init__(self, metadata: Mapping[str, bytes], wheel_name: str) -> None:
  39. self._metadata = metadata
  40. self._wheel_name = wheel_name
  41. def has_metadata(self, name: str) -> bool:
  42. return name in self._metadata
  43. def get_metadata(self, name: str) -> str:
  44. try:
  45. return self._metadata[name].decode()
  46. except UnicodeDecodeError as e:
  47. # Augment the default error with the origin of the file.
  48. raise UnsupportedWheel(
  49. f"Error decoding metadata for {self._wheel_name}: {e} in {name} file"
  50. )
  51. def get_metadata_lines(self, name: str) -> Iterable[str]:
  52. return pkg_resources.yield_lines(self.get_metadata(name))
  53. def metadata_isdir(self, name: str) -> bool:
  54. return False
  55. def metadata_listdir(self, name: str) -> list[str]:
  56. return []
  57. def run_script(self, script_name: str, namespace: str) -> None:
  58. pass
  59. class Distribution(BaseDistribution):
  60. def __init__(self, dist: pkg_resources.Distribution) -> None:
  61. self._dist = dist
  62. # This is populated lazily, to avoid loading metadata for all possible
  63. # distributions eagerly.
  64. self.__extra_mapping: Mapping[NormalizedName, str] | None = None
  65. @property
  66. def _extra_mapping(self) -> Mapping[NormalizedName, str]:
  67. if self.__extra_mapping is None:
  68. self.__extra_mapping = {
  69. canonicalize_name(extra): extra for extra in self._dist.extras
  70. }
  71. return self.__extra_mapping
  72. @classmethod
  73. def from_directory(cls, directory: str) -> BaseDistribution:
  74. dist_dir = directory.rstrip(os.sep)
  75. # Build a PathMetadata object, from path to metadata. :wink:
  76. base_dir, dist_dir_name = os.path.split(dist_dir)
  77. metadata = pkg_resources.PathMetadata(base_dir, dist_dir)
  78. # Determine the correct Distribution object type.
  79. if dist_dir.endswith(".egg-info"):
  80. dist_cls = pkg_resources.Distribution
  81. dist_name = os.path.splitext(dist_dir_name)[0]
  82. else:
  83. assert dist_dir.endswith(".dist-info")
  84. dist_cls = pkg_resources.DistInfoDistribution
  85. dist_name = os.path.splitext(dist_dir_name)[0].split("-")[0]
  86. dist = dist_cls(base_dir, project_name=dist_name, metadata=metadata)
  87. return cls(dist)
  88. @classmethod
  89. def from_metadata_file_contents(
  90. cls,
  91. metadata_contents: bytes,
  92. filename: str,
  93. project_name: str,
  94. ) -> BaseDistribution:
  95. metadata_dict = {
  96. "METADATA": metadata_contents,
  97. }
  98. dist = pkg_resources.DistInfoDistribution(
  99. location=filename,
  100. metadata=InMemoryMetadata(metadata_dict, filename),
  101. project_name=project_name,
  102. )
  103. return cls(dist)
  104. @classmethod
  105. def from_wheel(cls, wheel: Wheel, name: str) -> BaseDistribution:
  106. try:
  107. with wheel.as_zipfile() as zf:
  108. info_dir, _ = parse_wheel(zf, name)
  109. metadata_dict = {
  110. path.split("/", 1)[-1]: read_wheel_metadata_file(zf, path)
  111. for path in zf.namelist()
  112. if path.startswith(f"{info_dir}/")
  113. }
  114. except zipfile.BadZipFile as e:
  115. raise InvalidWheel(wheel.location, name) from e
  116. except UnsupportedWheel as e:
  117. raise UnsupportedWheel(f"{name} has an invalid wheel, {e}")
  118. dist = pkg_resources.DistInfoDistribution(
  119. location=wheel.location,
  120. metadata=InMemoryMetadata(metadata_dict, wheel.location),
  121. project_name=name,
  122. )
  123. return cls(dist)
  124. @property
  125. def location(self) -> str | None:
  126. return self._dist.location
  127. @property
  128. def installed_location(self) -> str | None:
  129. egg_link = egg_link_path_from_location(self.raw_name)
  130. if egg_link:
  131. location = egg_link
  132. elif self.location:
  133. location = self.location
  134. else:
  135. return None
  136. return normalize_path(location)
  137. @property
  138. def info_location(self) -> str | None:
  139. return self._dist.egg_info
  140. @property
  141. def installed_by_distutils(self) -> bool:
  142. # A distutils-installed distribution is provided by FileMetadata. This
  143. # provider has a "path" attribute not present anywhere else. Not the
  144. # best introspection logic, but pip has been doing this for a long time.
  145. try:
  146. return bool(self._dist._provider.path)
  147. except AttributeError:
  148. return False
  149. @property
  150. def canonical_name(self) -> NormalizedName:
  151. return canonicalize_name(self._dist.project_name)
  152. @property
  153. def version(self) -> Version:
  154. return parse_version(self._dist.version)
  155. @property
  156. def raw_version(self) -> str:
  157. return self._dist.version
  158. def is_file(self, path: InfoPath) -> bool:
  159. return self._dist.has_metadata(str(path))
  160. def iter_distutils_script_names(self) -> Iterator[str]:
  161. yield from self._dist.metadata_listdir("scripts")
  162. def read_text(self, path: InfoPath) -> str:
  163. name = str(path)
  164. if not self._dist.has_metadata(name):
  165. raise FileNotFoundError(name)
  166. content = self._dist.get_metadata(name)
  167. if content is None:
  168. raise NoneMetadataError(self, name)
  169. return content
  170. def iter_entry_points(self) -> Iterable[BaseEntryPoint]:
  171. for group, entries in self._dist.get_entry_map().items():
  172. for name, entry_point in entries.items():
  173. name, _, value = str(entry_point).partition("=")
  174. yield EntryPoint(name=name.strip(), value=value.strip(), group=group)
  175. def _metadata_impl(self) -> email.message.Message:
  176. """
  177. :raises NoneMetadataError: if the distribution reports `has_metadata()`
  178. True but `get_metadata()` returns None.
  179. """
  180. if isinstance(self._dist, pkg_resources.DistInfoDistribution):
  181. metadata_name = "METADATA"
  182. else:
  183. metadata_name = "PKG-INFO"
  184. try:
  185. metadata = self.read_text(metadata_name)
  186. except FileNotFoundError:
  187. if self.location:
  188. displaying_path = display_path(self.location)
  189. else:
  190. displaying_path = repr(self.location)
  191. logger.warning("No metadata found in %s", displaying_path)
  192. metadata = ""
  193. feed_parser = email.parser.FeedParser()
  194. feed_parser.feed(metadata)
  195. return feed_parser.close()
  196. def iter_dependencies(self, extras: Collection[str] = ()) -> Iterable[Requirement]:
  197. if extras:
  198. relevant_extras = set(self._extra_mapping) & set(
  199. map(canonicalize_name, extras)
  200. )
  201. extras = [self._extra_mapping[extra] for extra in relevant_extras]
  202. return self._dist.requires(extras)
  203. def iter_provided_extras(self) -> Iterable[NormalizedName]:
  204. return self._extra_mapping.keys()
  205. class Environment(BaseEnvironment):
  206. def __init__(self, ws: pkg_resources.WorkingSet) -> None:
  207. self._ws = ws
  208. @classmethod
  209. def default(cls) -> BaseEnvironment:
  210. return cls(pkg_resources.working_set)
  211. @classmethod
  212. def from_paths(cls, paths: list[str] | None) -> BaseEnvironment:
  213. return cls(pkg_resources.WorkingSet(paths))
  214. def _iter_distributions(self) -> Iterator[BaseDistribution]:
  215. for dist in self._ws:
  216. yield Distribution(dist)
  217. def _search_distribution(self, name: str) -> BaseDistribution | None:
  218. """Find a distribution matching the ``name`` in the environment.
  219. This searches from *all* distributions available in the environment, to
  220. match the behavior of ``pkg_resources.get_distribution()``.
  221. """
  222. canonical_name = canonicalize_name(name)
  223. for dist in self.iter_all_distributions():
  224. if dist.canonical_name == canonical_name:
  225. return dist
  226. return None
  227. def get_distribution(self, name: str) -> BaseDistribution | None:
  228. # Search the distribution by looking through the working set.
  229. dist = self._search_distribution(name)
  230. if dist:
  231. return dist
  232. # If distribution could not be found, call working_set.require to
  233. # update the working set, and try to find the distribution again.
  234. # This might happen for e.g. when you install a package twice, once
  235. # using setup.py develop and again using setup.py install. Now when
  236. # running pip uninstall twice, the package gets removed from the
  237. # working set in the first uninstall, so we have to populate the
  238. # working set again so that pip knows about it and the packages gets
  239. # picked up and is successfully uninstalled the second time too.
  240. try:
  241. # We didn't pass in any version specifiers, so this can never
  242. # raise pkg_resources.VersionConflict.
  243. self._ws.require(name)
  244. except pkg_resources.DistributionNotFound:
  245. return None
  246. return self._search_distribution(name)