pkg_resources.py 10 KB

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