Remove direct pkg_resources usages in req_install

This also by extension converts req_uninstall since the two are very
much coupled together.

Author: uranusjr

src/pip/_internal/commands/show.py

@@ -1,8 +1,6 @@
-import csv
 import logging
-import pathlib
 from optparse import Values
-from typing import Iterator, List, NamedTuple, Optional, Tuple
+from typing import Iterator, List, NamedTuple, Optional
 
 from pip._vendor.packaging.utils import canonicalize_name
 
@@ -69,33 +67,6 @@ class _PackageInfo(NamedTuple):
     files: Optional[List[str]]
 
 
-def _convert_legacy_entry(entry: Tuple[str, ...], info: Tuple[str, ...]) -> str:
-    """Convert a legacy installed-files.txt path into modern RECORD path.
-
-    The legacy format stores paths relative to the info directory, while the
-    modern format stores paths relative to the package root, e.g. the
-    site-packages directory.
-
-    :param entry: Path parts of the installed-files.txt entry.
-    :param info: Path parts of the egg-info directory relative to package root.
-    :returns: The converted entry.
-
-    For best compatibility with symlinks, this does not use ``abspath()`` or
-    ``Path.resolve()``, but tries to work with path parts:
-
-    1. While ``entry`` starts with ``..``, remove the equal amounts of parts
-       from ``info``; if ``info`` is empty, start appending ``..`` instead.
-    2. Join the two directly.
-    """
-    while entry and entry[0] == "..":
-        if not info or info[-1] == "..":
-            info += ("..",)
-        else:
-            info = info[:-1]
-        entry = entry[1:]
-    return str(pathlib.Path(*info, *entry))
-
-
 def search_packages_info(query: List[str]) -> Iterator[_PackageInfo]:
     """
     Gather details from installed distributions. Print distribution name,
@@ -121,34 +92,6 @@ def _get_requiring_packages(current_dist: BaseDistribution) -> List[str]:
             in {canonicalize_name(d.name) for d in dist.iter_dependencies()}
         ]
 
-    def _files_from_record(dist: BaseDistribution) -> Optional[Iterator[str]]:
-        try:
-            text = dist.read_text("RECORD")
-        except FileNotFoundError:
-            return None
-        # This extra Path-str cast normalizes entries.
-        return (str(pathlib.Path(row[0])) for row in csv.reader(text.splitlines()))
-
-    def _files_from_legacy(dist: BaseDistribution) -> Optional[Iterator[str]]:
-        try:
-            text = dist.read_text("installed-files.txt")
-        except FileNotFoundError:
-            return None
-        paths = (p for p in text.splitlines(keepends=False) if p)
-        root = dist.location
-        info = dist.info_directory
-        if root is None or info is None:
-            return paths
-        try:
-            info_rel = pathlib.Path(info).relative_to(root)
-        except ValueError:  # info is not relative to root.
-            return paths
-        if not info_rel.parts:  # info *is* root.
-            return paths
-        return (
-            _convert_legacy_entry(pathlib.Path(p).parts, info_rel.parts) for p in paths
-        )
-
     for query_name in query_names:
         try:
             dist = installed[query_name]
@@ -161,11 +104,11 @@ def _files_from_legacy(dist: BaseDistribution) -> Optional[Iterator[str]]:
         except FileNotFoundError:
             entry_points = []
 
-        files_iter = _files_from_record(dist) or _files_from_legacy(dist)
+        files_iter = dist.iter_files()
         if files_iter is None:
             files: Optional[List[str]] = None
         else:
-            files = sorted(files_iter)
+            files = sorted(str(f) for f in files_iter)
 
         metadata = dist.metadata
 

src/pip/_internal/distributions/sdist.py

@@ -19,9 +19,7 @@ class SourceDistribution(AbstractDistribution):
     """
 
     def get_metadata_distribution(self) -> BaseDistribution:
-        from pip._internal.metadata.pkg_resources import Distribution as _Dist
-
-        return _Dist(self.req.get_dist())
+        return self.req.get_dist()
 
     def prepare_distribution_metadata(
         self, finder: PackageFinder, build_isolation: bool

src/pip/_internal/distributions/wheel.py

@@ -5,7 +5,7 @@
 from pip._internal.metadata import (
     BaseDistribution,
     FilesystemWheel,
-    get_wheel_distribution,
+    get_distribution_for_wheel,
 )
 
 
@@ -23,7 +23,7 @@ def get_metadata_distribution(self) -> BaseDistribution:
         assert self.req.local_file_path, "Set as part of preparation during download"
         assert self.req.name, "Wheels are never unnamed"
         wheel = FilesystemWheel(self.req.local_file_path)
-        return get_wheel_distribution(wheel, canonicalize_name(self.req.name))
+        return get_distribution_for_wheel(wheel, canonicalize_name(self.req.name))
 
     def prepare_distribution_metadata(
         self, finder: PackageFinder, build_isolation: bool

src/pip/_internal/index/package_finder.py

@@ -881,7 +881,7 @@ def find_requirement(
 
         installed_version: Optional[_BaseVersion] = None
         if req.satisfied_by is not None:
-            installed_version = parse_version(req.satisfied_by.version)
+            installed_version = req.satisfied_by.version
 
         def _format_versions(cand_iter: Iterable[InstallationCandidate]) -> str:
             # This repeated parse_version and str() conversion is needed to

src/pip/_internal/metadata/__init__.py

@@ -38,7 +38,7 @@ def get_environment(paths: Optional[List[str]]) -> BaseEnvironment:
     return Environment.from_paths(paths)
 
 
-def get_wheel_distribution(wheel: Wheel, canonical_name: str) -> BaseDistribution:
+def get_distribution_for_wheel(wheel: Wheel, canonical_name: str) -> BaseDistribution:
     """Get the representation of the specified wheel's distribution metadata.
 
     This returns a Distribution instance from the chosen backend based on
@@ -49,3 +49,14 @@ def get_wheel_distribution(wheel: Wheel, canonical_name: str) -> BaseDistributio
     from .pkg_resources import Distribution
 
     return Distribution.from_wheel(wheel, canonical_name)
+
+
+def get_distribution_for_info_directory(directory_path: str) -> BaseDistribution:
+    """Get the specified info directory's distribution representation.
+
+    The directory should be an on-disk ``NAME-VERSION.dist-info`` or
+    ``NAME.egg-info`` directory.
+    """
+    from .pkg_resources import Distribution
+
+    return Distribution.from_info_directory(directory_path)

src/pip/_internal/metadata/base.py

@@ -1,6 +1,8 @@
+import csv
 import email.message
 import json
 import logging
+import pathlib
 import re
 import zipfile
 from typing import (
@@ -12,6 +14,7 @@
     Iterator,
     List,
     Optional,
+    Tuple,
     Union,
 )
 
@@ -38,6 +41,33 @@
 logger = logging.getLogger(__name__)
 
 
+def _convert_legacy_file(entry: Tuple[str, ...], info: Tuple[str, ...]) -> pathlib.Path:
+    """Convert a legacy installed-files.txt path into modern RECORD path.
+
+    The legacy format stores paths relative to the info directory, while the
+    modern format stores paths relative to the package root, e.g. the
+    site-packages directory.
+
+    :param entry: Path parts of the installed-files.txt entry.
+    :param info: Path parts of the egg-info directory relative to package root.
+    :returns: The converted entry.
+
+    For best compatibility with symlinks, this does not use ``abspath()`` or
+    ``Path.resolve()``, but tries to work with path parts:
+
+    1. While ``entry`` starts with ``..``, remove the equal amounts of parts
+       from ``info``; if ``info`` is empty, start appending ``..`` instead.
+    2. Join the two directly.
+    """
+    while entry and entry[0] == "..":
+        if not info or info[-1] == "..":
+            info += ("..",)
+        else:
+            info = info[:-1]
+        entry = entry[1:]
+    return pathlib.Path(*info, *entry)
+
+
 class BaseEntryPoint(Protocol):
     @property
     def name(self) -> str:
@@ -127,6 +157,17 @@ def direct_url(self) -> Optional[DirectUrl]:
     def installer(self) -> str:
         raise NotImplementedError()
 
+    @property
+    def egg_link(self) -> Optional[str]:
+        """Location of the ``.egg-link`` for this distribution.
+
+        If there's not a matching file, None is returned. Note that finding this
+        file does not necessarily mean the currently-installed distribution is
+        editable since the ``.egg-link`` can still be shadowed by a
+        non-editable installation located in front of it in ``sys.path``.
+        """
+        raise NotImplementedError()
+
     @property
     def editable(self) -> bool:
         raise NotImplementedError()
@@ -146,11 +187,20 @@ def in_site_packages(self) -> bool:
     def read_text(self, name: str) -> str:
         """Read a file in the .dist-info (or .egg-info) directory.
 
-        Should raise ``FileNotFoundError`` if ``name`` does not exist in the
-        metadata directory.
+        :raises FileNotFoundError: ``name`` does not exist in the info directory.
         """
         raise NotImplementedError()
 
+    def iterdir(self, name: str) -> Iterable[pathlib.PurePosixPath]:
+        """Iterate through a directory in the info directory.
+
+        Each item is a path relative to the info directory.
+
+        :raises FileNotFoundError: ``name`` does not exist in the info directory.
+        :raises NotADirectoryError: ``name`` exists in the info directory, but
+            is not a directory.
+        """
+
     def iter_entry_points(self) -> Iterable[BaseEntryPoint]:
         raise NotImplementedError()
 
@@ -206,6 +256,43 @@ def iter_provided_extras(self) -> Iterable[str]:
         """
         raise NotImplementedError()
 
+    def _iter_files_from_legacy(self) -> Optional[Iterator[pathlib.Path]]:
+        try:
+            text = self.read_text("installed-files.txt")
+        except FileNotFoundError:
+            return None
+        paths = (pathlib.Path(p) for p in text.splitlines(keepends=False) if p)
+        root = self.location
+        info = self.info_directory
+        if root is None or info is None:
+            return paths
+        try:
+            rel = pathlib.Path(info).relative_to(root)
+        except ValueError:  # info is not relative to root.
+            return paths
+        if not rel.parts:  # info *is* root.
+            return paths
+        return (_convert_legacy_file(p.parts, rel.parts) for p in paths)
+
+    def _iter_files_from_record(self) -> Optional[Iterator[pathlib.Path]]:
+        try:
+            text = self.read_text("RECORD")
+        except FileNotFoundError:
+            return None
+        return (pathlib.Path(row[0]) for row in csv.reader(text.splitlines()))
+
+    def iter_files(self) -> Optional[Iterator[pathlib.Path]]:
+        """Files in the distribution's record.
+
+        For modern .dist-info distributions, this is the files listed in the
+        ``RECORD`` file. All entries are paths relative to this distribution's
+        ``location``.
+
+        Note that this can be None for unmanagable distributions, e.g. an
+        installation performed by distutils or a foreign package manager.
+        """
+        return self._iter_files_from_record() or self._iter_files_from_legacy()
+
 
 class BaseEnvironment:
     """An environment containing distributions to introspect."""

src/pip/_internal/metadata/pkg_resources.py

@@ -1,5 +1,7 @@
 import email.message
 import logging
+import os
+import pathlib
 from typing import (
     TYPE_CHECKING,
     Collection,
@@ -49,6 +51,25 @@ def from_wheel(cls, wheel: Wheel, name: str) -> "Distribution":
             dist = pkg_resources_distribution_for_wheel(zf, name, wheel.location)
         return cls(dist)
 
+    @classmethod
+    def from_info_directory(cls, path: str) -> "Distribution":
+        dist_dir = path.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("-", 1)[0]
+
+        return cls(dist_cls(base_dir, project_name=dist_name, metadata=metadata))
+
     @property
     def location(self) -> Optional[str]:
         return self._dist.location
@@ -63,12 +84,19 @@ def canonical_name(self) -> "NormalizedName":
 
     @property
     def version(self) -> DistributionVersion:
+        # pkg_resouces may contain a different copy of packaging.version from
+        # pip in if the downstream distributor does a poor job debundling pip.
+        # We avoid parsed_version and use our vendored packaging instead.
         return parse_version(self._dist.version)
 
     @property
     def installer(self) -> str:
         return get_installer(self._dist)
 
+    @property
+    def egg_link(self) -> Optional[str]:
+        return misc.egg_link_path(self._dist)
+
     @property
     def editable(self) -> bool:
         return misc.dist_is_editable(self._dist)
@@ -90,6 +118,15 @@ def read_text(self, name: str) -> str:
             raise FileNotFoundError(name)
         return self._dist.get_metadata(name)
 
+    def iterdir(self, name: str) -> Iterable[pathlib.PurePosixPath]:
+        if not self._dist.has_metadata(name):
+            raise FileNotFoundError(name)
+        if not self._dist.metadata_isdir(name):
+            raise NotADirectoryError(name)
+        return (
+            pathlib.PurePosixPath(name, n) for n in self._dist.metadata_listdir(name)
+        )
+
     def iter_entry_points(self) -> Iterable[BaseEntryPoint]:
         for group, entries in self._dist.get_entry_map().items():
             for name, entry_point in entries.items():

src/pip/_internal/network/lazy_wheel.py

@@ -11,7 +11,11 @@
 from pip._vendor.packaging.utils import canonicalize_name
 from pip._vendor.requests.models import CONTENT_CHUNK_SIZE, Response
 
-from pip._internal.metadata import BaseDistribution, MemoryWheel, get_wheel_distribution
+from pip._internal.metadata import (
+    BaseDistribution,
+    MemoryWheel,
+    get_distribution_for_wheel,
+)
 from pip._internal.network.session import PipSession
 from pip._internal.network.utils import HEADERS, raise_for_status, response_chunks
 
@@ -34,7 +38,7 @@ def dist_from_wheel_url(name: str, url: str, session: PipSession) -> BaseDistrib
         wheel = MemoryWheel(zf.name, zf)  # type: ignore
         # After context manager exit, wheel.name
         # is an invalid file by intention.
-        return get_wheel_distribution(wheel, canonicalize_name(name))
+        return get_distribution_for_wheel(wheel, canonicalize_name(name))
 
 
 class LazyZipOverHTTP: