Replace more pkg_resources usages

news/10157.process.rst

@@ -0,0 +1,7 @@
+``pip freeze``, ``pip list``, and ``pip show`` no longer normalize underscore
+(``_``) in distribution names to dash (``-``). This is a side effect of the
+migration to ``importlib.metadata``, since the underscore-dash normalization
+behavior is non-standard and specific to setuptools. This should not affect
+other parts of pip (for example, when feeding the ``pip freeze`` result back
+into ``pip install``) since pip internally performs standard PEP 503
+normalization independently to setuptools.

src/pip/_internal/build_env.py

@@ -14,11 +14,13 @@
 from typing import TYPE_CHECKING, Iterable, Iterator, List, Optional, Set, Tuple, Type
 
 from pip._vendor.certifi import where
-from pip._vendor.pkg_resources import Requirement, VersionConflict, WorkingSet
+from pip._vendor.packaging.requirements import Requirement
+from pip._vendor.packaging.version import Version
 
 from pip import __file__ as pip_location
 from pip._internal.cli.spinners import open_spinner
 from pip._internal.locations import get_platlib, get_prefixed_libs, get_purelib
+from pip._internal.metadata import get_environment
 from pip._internal.utils.subprocess import call_subprocess
 from pip._internal.utils.temp_dir import TempDirectory, tempdir_kinds
 
@@ -167,14 +169,20 @@ def check_requirements(self, reqs):
         missing = set()
         conflicting = set()
         if reqs:
-            ws = WorkingSet(self._lib_dirs)
-            for req in reqs:
-                try:
-                    if ws.find(Requirement.parse(req)) is None:
-                        missing.add(req)
-                except VersionConflict as e:
-                    conflicting.add((str(e.args[0].as_requirement()),
-                                     str(e.args[1])))
+            env = get_environment(self._lib_dirs)
+            for req_str in reqs:
+                req = Requirement(req_str)
+                dist = env.get_distribution(req.name)
+                if not dist:
+                    missing.add(req_str)
+                    continue
+                if isinstance(dist.version, Version):
+                    installed_req_str = f"{req.name}=={dist.version}"
+                else:
+                    installed_req_str = f"{req.name}==={dist.version}"
+                if dist.version not in req.specifier:
+                    conflicting.add((installed_req_str, req_str))
+                # FIXME: Consider direct URL?
         return conflicting, missing
 
     def install_requirements(

src/pip/_internal/cli/autocompletion.py

@@ -9,7 +9,7 @@
 
 from pip._internal.cli.main_parser import create_main_parser
 from pip._internal.commands import commands_dict, create_command
-from pip._internal.utils.misc import get_installed_distributions
+from pip._internal.metadata import get_default_environment
 
 
 def autocomplete() -> None:
@@ -45,11 +45,13 @@ def autocomplete() -> None:
             "uninstall",
         ]
         if should_list_installed:
+            env = get_default_environment()
             lc = current.lower()
             installed = [
-                dist.key
-                for dist in get_installed_distributions(local_only=True)
-                if dist.key.startswith(lc) and dist.key not in cwords[1:]
+                dist.canonical_name
+                for dist in env.iter_installed_distributions(local_only=True)
+                if dist.canonical_name.startswith(lc)
+                and dist.canonical_name not in cwords[1:]
             ]
             # if there are no dists installed, fall back to option completion
             if installed:

src/pip/_internal/commands/list.py

@@ -152,7 +152,7 @@ def run(self, options, args):
 
         skip = set(stdlib_pkgs)
         if options.excludes:
-            skip.update(options.excludes)
+            skip.update(canonicalize_name(n) for n in options.excludes)
 
         packages: "_ProcessedDists" = [
             cast("_DistWithLatestInfo", d)
@@ -199,7 +199,7 @@ def get_not_required(self, packages, options):
         dep_keys = {
             canonicalize_name(dep.name)
             for dist in packages
-            for dep in dist.iter_dependencies()
+            for dep in (dist.iter_dependencies() or ())
         }
 
         # Create a set to remove duplicate packages, and cast it to a list
@@ -252,10 +252,10 @@ def output_package_listing(self, packages, options):
         elif options.list_format == 'freeze':
             for dist in packages:
                 if options.verbose >= 1:
-                    write_output("%s==%s (%s)", dist.canonical_name,
+                    write_output("%s==%s (%s)", dist.raw_name,
                                  dist.version, dist.location)
                 else:
-                    write_output("%s==%s", dist.canonical_name, dist.version)
+                    write_output("%s==%s", dist.raw_name, dist.version)
         elif options.list_format == 'json':
             write_output(format_for_json(packages, options))
 
@@ -297,7 +297,7 @@ def format_for_columns(pkgs, options):
     for proj in pkgs:
         # if we're working on the 'outdated' list, separate out the
         # latest_version and type
-        row = [proj.canonical_name, str(proj.version)]
+        row = [proj.raw_name, str(proj.version)]
 
         if running_outdated:
             row.append(str(proj.latest_version))
@@ -318,7 +318,7 @@ def format_for_json(packages, options):
     data = []
     for dist in packages:
         info = {
-            'name': dist.canonical_name,
+            'name': dist.raw_name,
             'version': str(dist.version),
         }
         if options.verbose >= 1:

src/pip/_internal/commands/show.py

@@ -1,14 +1,14 @@
+import csv
 import logging
 import os
-from email.parser import FeedParser
 from optparse import Values
-from typing import Dict, Iterator, List
+from typing import Iterator, List, NamedTuple, Optional
 
-from pip._vendor import pkg_resources
 from pip._vendor.packaging.utils import canonicalize_name
 
 from pip._internal.cli.base_command import Command
 from pip._internal.cli.status_codes import ERROR, SUCCESS
+from pip._internal.metadata import BaseDistribution, get_default_environment
 from pip._internal.utils.misc import write_output
 
 logger = logging.getLogger(__name__)
@@ -50,98 +50,111 @@ def run(self, options, args):
         return SUCCESS
 
 
-def search_packages_info(query):
-    # type: (List[str]) -> Iterator[Dict[str, str]]
+class _PackageInfo(NamedTuple):
+    name: str
+    version: str
+    location: str
+    requires: List[str]
+    required_by: List[str]
+    installer: str
+    metadata_version: str
+    classifiers: List[str]
+    summary: str
+    homepage: str
+    author: str
+    author_email: str
+    license: str
+    entry_points: List[str]
+    files: Optional[List[str]]
+
+
+def search_packages_info(query: List[str]) -> Iterator[_PackageInfo]:
     """
     Gather details from installed distributions. Print distribution name,
     version, location, and installed files. Installed files requires a
     pip generated 'installed-files.txt' in the distributions '.egg-info'
     directory.
     """
-    installed = {}
-    for p in pkg_resources.working_set:
-        installed[canonicalize_name(p.project_name)] = p
+    env = get_default_environment()
 
+    installed = {
+        dist.canonical_name: dist
+        for dist in env.iter_distributions()
+    }
     query_names = [canonicalize_name(name) for name in query]
     missing = sorted(
         [name for name, pkg in zip(query, query_names) if pkg not in installed]
     )
     if missing:
         logger.warning('Package(s) not found: %s', ', '.join(missing))
 
-    def get_requiring_packages(package_name):
-        # type: (str) -> List[str]
-        canonical_name = canonicalize_name(package_name)
+    def _get_requiring_packages(current_dist: BaseDistribution) -> List[str]:
         return [
-            pkg.project_name for pkg in pkg_resources.working_set
-            if canonical_name in
-               [canonicalize_name(required.name) for required in
-                pkg.requires()]
+            dist.metadata["Name"] or "UNKNOWN"
+            for dist in installed.values()
+            if current_dist.canonical_name in {
+                canonicalize_name(d.name) for d in dist.iter_dependencies()
+            }
         ]
 
-    for dist in [installed[pkg] for pkg in query_names if pkg in installed]:
-        package = {
-            'name': dist.project_name,
-            'version': dist.version,
-            'location': dist.location,
-            'requires': [dep.project_name for dep in dist.requires()],
-            'required_by': get_requiring_packages(dist.project_name)
-        }
-        file_list = None
-        metadata = ''
-        if isinstance(dist, pkg_resources.DistInfoDistribution):
-            # RECORDs should be part of .dist-info metadatas
-            if dist.has_metadata('RECORD'):
-                lines = dist.get_metadata_lines('RECORD')
-                paths = [line.split(',')[0] for line in lines]
-                paths = [os.path.join(dist.location, p) for p in paths]
-                file_list = [os.path.relpath(p, dist.location) for p in paths]
-
-            if dist.has_metadata('METADATA'):
-                metadata = dist.get_metadata('METADATA')
+    def _files_from_record(dist: BaseDistribution) -> Optional[Iterator[str]]:
+        try:
+            text = dist.read_text('RECORD')
+        except FileNotFoundError:
+            return None
+        return (row[0] for row in csv.reader(text.splitlines()))
+
+    def _files_from_installed_files(dist: BaseDistribution) -> Optional[Iterator[str]]:
+        try:
+            text = dist.read_text('installed-files.txt')
+        except FileNotFoundError:
+            return None
+        return (p for p in text.splitlines(keepends=False) if p)
+
+    for query_name in query_names:
+        try:
+            dist = installed[query_name]
+        except KeyError:
+            continue
+
+        try:
+            entry_points_text = dist.read_text('entry_points.txt')
+            entry_points = entry_points_text.splitlines(keepends=False)
+        except FileNotFoundError:
+            entry_points = []
+
+        files_iter = _files_from_record(dist) or _files_from_installed_files(dist)
+        if files_iter is None:
+            files: Optional[List[str]] = None
         else:
-            # Otherwise use pip's log for .egg-info's
-            if dist.has_metadata('installed-files.txt'):
-                paths = dist.get_metadata_lines('installed-files.txt')
-                paths = [os.path.join(dist.egg_info, p) for p in paths]
-                file_list = [os.path.relpath(p, dist.location) for p in paths]
-
-            if dist.has_metadata('PKG-INFO'):
-                metadata = dist.get_metadata('PKG-INFO')
-
-        if dist.has_metadata('entry_points.txt'):
-            entry_points = dist.get_metadata_lines('entry_points.txt')
-            package['entry_points'] = entry_points
-
-        if dist.has_metadata('INSTALLER'):
-            for line in dist.get_metadata_lines('INSTALLER'):
-                if line.strip():
-                    package['installer'] = line.strip()
-                    break
-
-        # @todo: Should pkg_resources.Distribution have a
-        # `get_pkg_info` method?
-        feed_parser = FeedParser()
-        feed_parser.feed(metadata)
-        pkg_info_dict = feed_parser.close()
-        for key in ('metadata-version', 'summary',
-                    'home-page', 'author', 'author-email', 'license'):
-            package[key] = pkg_info_dict.get(key)
-
-        # It looks like FeedParser cannot deal with repeated headers
-        classifiers = []
-        for line in metadata.splitlines():
-            if line.startswith('Classifier: '):
-                classifiers.append(line[len('Classifier: '):])
-        package['classifiers'] = classifiers
-
-        if file_list:
-            package['files'] = sorted(file_list)
-        yield package
-
-
-def print_results(distributions, list_files=False, verbose=False):
-    # type: (Iterator[Dict[str, str]], bool, bool) -> bool
+            files = sorted(os.path.relpath(p, dist.location) for p in files_iter)
+
+        metadata = dist.metadata
+
+        yield _PackageInfo(
+            name=dist.raw_name,
+            version=str(dist.version),
+            location=dist.location or "",
+            requires=[req.name for req in dist.iter_dependencies()],
+            required_by=_get_requiring_packages(dist),
+            installer=dist.installer,
+            metadata_version=dist.metadata_version or "",
+            classifiers=metadata.get_all("Classifier", []),
+            summary=metadata.get("Summary", ""),
+            homepage=metadata.get("Home-page", ""),
+            author=metadata.get("Author", ""),
+            author_email=metadata.get("Author-email", ""),
+            license=metadata.get("License", ""),
+            entry_points=entry_points,
+            files=files,
+        )
+
+
+def print_results(
+    distributions: Iterator[_PackageInfo],
+    list_files: bool,
+    verbose: bool,
+) -> bool:
     """
     Print the information from installed distributions found.
     """
@@ -151,31 +164,31 @@ def print_results(distributions, list_files=False, verbose=False):
         if i > 0:
             write_output("---")
 
-        write_output("Name: %s", dist.get('name', ''))
-        write_output("Version: %s", dist.get('version', ''))
-        write_output("Summary: %s", dist.get('summary', ''))
-        write_output("Home-page: %s", dist.get('home-page', ''))
-        write_output("Author: %s", dist.get('author', ''))
-        write_output("Author-email: %s", dist.get('author-email', ''))
-        write_output("License: %s", dist.get('license', ''))
-        write_output("Location: %s", dist.get('location', ''))
-        write_output("Requires: %s", ', '.join(dist.get('requires', [])))
-        write_output("Required-by: %s", ', '.join(dist.get('required_by', [])))
+        write_output("Name: %s", dist.name)
+        write_output("Version: %s", dist.version)
+        write_output("Summary: %s", dist.summary)
+        write_output("Home-page: %s", dist.homepage)
+        write_output("Author: %s", dist.author)
+        write_output("Author-email: %s", dist.author_email)
+        write_output("License: %s", dist.license)
+        write_output("Location: %s", dist.location)
+        write_output("Requires: %s", ', '.join(dist.requires))
+        write_output("Required-by: %s", ', '.join(dist.required_by))
 
         if verbose:
-            write_output("Metadata-Version: %s",
-                         dist.get('metadata-version', ''))
-            write_output("Installer: %s", dist.get('installer', ''))
+            write_output("Metadata-Version: %s", dist.metadata_version)
+            write_output("Installer: %s", dist.installer)
             write_output("Classifiers:")
-            for classifier in dist.get('classifiers', []):
+            for classifier in dist.classifiers:
                 write_output("  %s", classifier)
             write_output("Entry-points:")
-            for entry in dist.get('entry_points', []):
+            for entry in dist.entry_points:
                 write_output("  %s", entry.strip())
         if list_files:
             write_output("Files:")
-            for line in dist.get('files', []):
-                write_output("  %s", line.strip())
-            if "files" not in dist:
-                write_output("Cannot locate installed-files.txt")
+            if dist.files is None:
+                write_output("Cannot locate RECORD or installed-files.txt")
+            else:
+                for line in dist.files:
+                    write_output("  %s", line.strip())
     return results_printed

src/pip/_internal/metadata/__init__.py

@@ -2,6 +2,14 @@
 
 from .base import BaseDistribution, BaseEnvironment
 
+__all__ = [
+    "BaseDistribution",
+    "BaseEnvironment",
+    "get_default_environment",
+    "get_environment",
+    "get_wheel_distribution",
+]
+
 
 def get_default_environment() -> BaseEnvironment:
     """Get the default representation for the current environment.