| import email.message |
| import email.parser |
| import logging |
| import os |
| import zipfile |
| from typing import Collection, Iterable, Iterator, List, Mapping, NamedTuple, Optional |
| |
| from pip._vendor import pkg_resources |
| from pip._vendor.packaging.requirements import Requirement |
| from pip._vendor.packaging.utils import NormalizedName, canonicalize_name |
| from pip._vendor.packaging.version import parse as parse_version |
| |
| from pip._internal.exceptions import InvalidWheel, NoneMetadataError, UnsupportedWheel |
| from pip._internal.utils.egg_link import egg_link_path_from_location |
| from pip._internal.utils.misc import display_path, normalize_path |
| from pip._internal.utils.wheel import parse_wheel, read_wheel_metadata_file |
| |
| from .base import ( |
| BaseDistribution, |
| BaseEntryPoint, |
| BaseEnvironment, |
| DistributionVersion, |
| InfoPath, |
| Wheel, |
| ) |
| |
| logger = logging.getLogger(__name__) |
| |
| |
| class EntryPoint(NamedTuple): |
| name: str |
| value: str |
| group: str |
| |
| |
| class InMemoryMetadata: |
| """IMetadataProvider that reads metadata files from a dictionary. |
| |
| This also maps metadata decoding exceptions to our internal exception type. |
| """ |
| |
| def __init__(self, metadata: Mapping[str, bytes], wheel_name: str) -> None: |
| self._metadata = metadata |
| self._wheel_name = wheel_name |
| |
| def has_metadata(self, name: str) -> bool: |
| return name in self._metadata |
| |
| def get_metadata(self, name: str) -> str: |
| try: |
| return self._metadata[name].decode() |
| except UnicodeDecodeError as e: |
| # Augment the default error with the origin of the file. |
| raise UnsupportedWheel( |
| f"Error decoding metadata for {self._wheel_name}: {e} in {name} file" |
| ) |
| |
| def get_metadata_lines(self, name: str) -> Iterable[str]: |
| return pkg_resources.yield_lines(self.get_metadata(name)) |
| |
| def metadata_isdir(self, name: str) -> bool: |
| return False |
| |
| def metadata_listdir(self, name: str) -> List[str]: |
| return [] |
| |
| def run_script(self, script_name: str, namespace: str) -> None: |
| pass |
| |
| |
| class Distribution(BaseDistribution): |
| def __init__(self, dist: pkg_resources.Distribution) -> None: |
| self._dist = dist |
| |
| @classmethod |
| def from_directory(cls, directory: str) -> BaseDistribution: |
| dist_dir = directory.rstrip(os.sep) |
| |
| # Build a PathMetadata object, from path to metadata. :wink: |
| base_dir, dist_dir_name = os.path.split(dist_dir) |
| metadata = pkg_resources.PathMetadata(base_dir, dist_dir) |
| |
| # Determine the correct Distribution object type. |
| if dist_dir.endswith(".egg-info"): |
| dist_cls = pkg_resources.Distribution |
| dist_name = os.path.splitext(dist_dir_name)[0] |
| else: |
| assert dist_dir.endswith(".dist-info") |
| dist_cls = pkg_resources.DistInfoDistribution |
| dist_name = os.path.splitext(dist_dir_name)[0].split("-")[0] |
| |
| dist = dist_cls(base_dir, project_name=dist_name, metadata=metadata) |
| return cls(dist) |
| |
| @classmethod |
| def from_metadata_file_contents( |
| cls, |
| metadata_contents: bytes, |
| filename: str, |
| project_name: str, |
| ) -> BaseDistribution: |
| metadata_dict = { |
| "METADATA": metadata_contents, |
| } |
| dist = pkg_resources.DistInfoDistribution( |
| location=filename, |
| metadata=InMemoryMetadata(metadata_dict, filename), |
| project_name=project_name, |
| ) |
| return cls(dist) |
| |
| @classmethod |
| def from_wheel(cls, wheel: Wheel, name: str) -> BaseDistribution: |
| try: |
| with wheel.as_zipfile() as zf: |
| info_dir, _ = parse_wheel(zf, name) |
| metadata_dict = { |
| path.split("/", 1)[-1]: read_wheel_metadata_file(zf, path) |
| for path in zf.namelist() |
| if path.startswith(f"{info_dir}/") |
| } |
| except zipfile.BadZipFile as e: |
| raise InvalidWheel(wheel.location, name) from e |
| except UnsupportedWheel as e: |
| raise UnsupportedWheel(f"{name} has an invalid wheel, {e}") |
| dist = pkg_resources.DistInfoDistribution( |
| location=wheel.location, |
| metadata=InMemoryMetadata(metadata_dict, wheel.location), |
| project_name=name, |
| ) |
| return cls(dist) |
| |
| @property |
| def location(self) -> Optional[str]: |
| return self._dist.location |
| |
| @property |
| def installed_location(self) -> Optional[str]: |
| egg_link = egg_link_path_from_location(self.raw_name) |
| if egg_link: |
| location = egg_link |
| elif self.location: |
| location = self.location |
| else: |
| return None |
| return normalize_path(location) |
| |
| @property |
| def info_location(self) -> Optional[str]: |
| return self._dist.egg_info |
| |
| @property |
| def installed_by_distutils(self) -> bool: |
| # A distutils-installed distribution is provided by FileMetadata. This |
| # provider has a "path" attribute not present anywhere else. Not the |
| # best introspection logic, but pip has been doing this for a long time. |
| try: |
| return bool(self._dist._provider.path) |
| except AttributeError: |
| return False |
| |
| @property |
| def canonical_name(self) -> NormalizedName: |
| return canonicalize_name(self._dist.project_name) |
| |
| @property |
| def version(self) -> DistributionVersion: |
| return parse_version(self._dist.version) |
| |
| def is_file(self, path: InfoPath) -> bool: |
| return self._dist.has_metadata(str(path)) |
| |
| def iter_distutils_script_names(self) -> Iterator[str]: |
| yield from self._dist.metadata_listdir("scripts") |
| |
| def read_text(self, path: InfoPath) -> str: |
| name = str(path) |
| if not self._dist.has_metadata(name): |
| raise FileNotFoundError(name) |
| content = self._dist.get_metadata(name) |
| if content is None: |
| raise NoneMetadataError(self, name) |
| return content |
| |
| def iter_entry_points(self) -> Iterable[BaseEntryPoint]: |
| for group, entries in self._dist.get_entry_map().items(): |
| for name, entry_point in entries.items(): |
| name, _, value = str(entry_point).partition("=") |
| yield EntryPoint(name=name.strip(), value=value.strip(), group=group) |
| |
| def _metadata_impl(self) -> email.message.Message: |
| """ |
| :raises NoneMetadataError: if the distribution reports `has_metadata()` |
| True but `get_metadata()` returns None. |
| """ |
| if isinstance(self._dist, pkg_resources.DistInfoDistribution): |
| metadata_name = "METADATA" |
| else: |
| metadata_name = "PKG-INFO" |
| try: |
| metadata = self.read_text(metadata_name) |
| except FileNotFoundError: |
| if self.location: |
| displaying_path = display_path(self.location) |
| else: |
| displaying_path = repr(self.location) |
| logger.warning("No metadata found in %s", displaying_path) |
| metadata = "" |
| feed_parser = email.parser.FeedParser() |
| feed_parser.feed(metadata) |
| return feed_parser.close() |
| |
| def iter_dependencies(self, extras: Collection[str] = ()) -> Iterable[Requirement]: |
| if extras: # pkg_resources raises on invalid extras, so we sanitize. |
| extras = frozenset(extras).intersection(self._dist.extras) |
| return self._dist.requires(extras) |
| |
| def iter_provided_extras(self) -> Iterable[str]: |
| return self._dist.extras |
| |
| |
| class Environment(BaseEnvironment): |
| def __init__(self, ws: pkg_resources.WorkingSet) -> None: |
| self._ws = ws |
| |
| @classmethod |
| def default(cls) -> BaseEnvironment: |
| return cls(pkg_resources.working_set) |
| |
| @classmethod |
| def from_paths(cls, paths: Optional[List[str]]) -> BaseEnvironment: |
| return cls(pkg_resources.WorkingSet(paths)) |
| |
| def _iter_distributions(self) -> Iterator[BaseDistribution]: |
| for dist in self._ws: |
| yield Distribution(dist) |
| |
| def _search_distribution(self, name: str) -> Optional[BaseDistribution]: |
| """Find a distribution matching the ``name`` in the environment. |
| |
| This searches from *all* distributions available in the environment, to |
| match the behavior of ``pkg_resources.get_distribution()``. |
| """ |
| canonical_name = canonicalize_name(name) |
| for dist in self.iter_all_distributions(): |
| if dist.canonical_name == canonical_name: |
| return dist |
| return None |
| |
| def get_distribution(self, name: str) -> Optional[BaseDistribution]: |
| # Search the distribution by looking through the working set. |
| dist = self._search_distribution(name) |
| if dist: |
| return dist |
| |
| # If distribution could not be found, call working_set.require to |
| # update the working set, and try to find the distribution again. |
| # This might happen for e.g. when you install a package twice, once |
| # using setup.py develop and again using setup.py install. Now when |
| # running pip uninstall twice, the package gets removed from the |
| # working set in the first uninstall, so we have to populate the |
| # working set again so that pip knows about it and the packages gets |
| # picked up and is successfully uninstalled the second time too. |
| try: |
| # We didn't pass in any version specifiers, so this can never |
| # raise pkg_resources.VersionConflict. |
| self._ws.require(name) |
| except pkg_resources.DistributionNotFound: |
| return None |
| return self._search_distribution(name) |
