Remove direct pkg_resources usages in req_install and req_uninstall

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: