Migrate 'pip list' to use metadata abstraction

src/pip/_internal/commands/list.py

@@ -1,28 +1,37 @@
 import json
 import logging
 from optparse import Values
-from typing import Iterator, List, Set, Tuple
+from typing import TYPE_CHECKING, Iterator, List, Optional, Sequence, Tuple, cast
 
-from pip._vendor.pkg_resources import Distribution
+from pip._vendor.packaging.utils import canonicalize_name
 
 from pip._internal.cli import cmdoptions
 from pip._internal.cli.req_command import IndexGroupCommand
 from pip._internal.cli.status_codes import SUCCESS
 from pip._internal.exceptions import CommandError
 from pip._internal.index.collector import LinkCollector
 from pip._internal.index.package_finder import PackageFinder
+from pip._internal.metadata import BaseDistribution, get_environment
 from pip._internal.models.selection_prefs import SelectionPreferences
 from pip._internal.network.session import PipSession
-from pip._internal.utils.compat import stdlib_pkgs
-from pip._internal.utils.misc import (
-    dist_is_editable,
-    get_installed_distributions,
-    tabulate,
-    write_output,
-)
-from pip._internal.utils.packaging import get_installer
+from pip._internal.utils.misc import stdlib_pkgs, tabulate, write_output
 from pip._internal.utils.parallel import map_multithread
 
+if TYPE_CHECKING:
+    from pip._internal.metadata.base import DistributionVersion
+
+    class _DistWithLatestInfo(BaseDistribution):
+        """Give the distribution object a couple of extra fields.
+
+        These will be populated during ``get_outdated()``. This is dirty but
+        makes the rest of the code much cleaner.
+        """
+        latest_version: DistributionVersion
+        latest_filetype: str
+
+    _ProcessedDists = Sequence[_DistWithLatestInfo]
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -145,14 +154,16 @@ def run(self, options, args):
         if options.excludes:
             skip.update(options.excludes)
 
-        packages = get_installed_distributions(
-            local_only=options.local,
-            user_only=options.user,
-            editables_only=options.editable,
-            include_editables=options.include_editable,
-            paths=options.path,
-            skip=skip,
-        )
+        packages: "_ProcessedDists" = [
+            cast("_DistWithLatestInfo", d)
+            for d in get_environment(options.path).iter_installed_distributions(
+                local_only=options.local,
+                user_only=options.user,
+                editables_only=options.editable,
+                include_editables=options.include_editable,
+                skip=skip,
+            )
+        ]
 
         # get_not_required must be called firstly in order to find and
         # filter out all dependencies correctly. Otherwise a package
@@ -170,45 +181,47 @@ def run(self, options, args):
         return SUCCESS
 
     def get_outdated(self, packages, options):
-        # type: (List[Distribution], Values) -> List[Distribution]
+        # type: (_ProcessedDists, Values) -> _ProcessedDists
         return [
             dist for dist in self.iter_packages_latest_infos(packages, options)
-            if dist.latest_version > dist.parsed_version
+            if dist.latest_version > dist.version
         ]
 
     def get_uptodate(self, packages, options):
-        # type: (List[Distribution], Values) -> List[Distribution]
+        # type: (_ProcessedDists, Values) -> _ProcessedDists
         return [
             dist for dist in self.iter_packages_latest_infos(packages, options)
-            if dist.latest_version == dist.parsed_version
+            if dist.latest_version == dist.version
         ]
 
     def get_not_required(self, packages, options):
-        # type: (List[Distribution], Values) -> List[Distribution]
-        dep_keys = set()  # type: Set[Distribution]
-        for dist in packages:
-            dep_keys.update(requirement.key for requirement in dist.requires())
+        # type: (_ProcessedDists, Values) -> _ProcessedDists
+        dep_keys = {
+            canonicalize_name(dep.name)
+            for dist in packages
+            for dep in dist.iter_dependencies()
+        }
 
         # Create a set to remove duplicate packages, and cast it to a list
         # to keep the return type consistent with get_outdated and
         # get_uptodate
-        return list({pkg for pkg in packages if pkg.key not in dep_keys})
+        return list({pkg for pkg in packages if pkg.canonical_name not in dep_keys})
 
     def iter_packages_latest_infos(self, packages, options):
-        # type: (List[Distribution], Values) -> Iterator[Distribution]
+        # type: (_ProcessedDists, Values) -> Iterator[_DistWithLatestInfo]
         with self._build_session(options) as session:
             finder = self._build_package_finder(options, session)
 
             def latest_info(dist):
-                # type: (Distribution) -> Distribution
-                all_candidates = finder.find_all_candidates(dist.key)
+                # type: (_DistWithLatestInfo) -> Optional[_DistWithLatestInfo]
+                all_candidates = finder.find_all_candidates(dist.canonical_name)
                 if not options.pre:
                     # Remove prereleases
                     all_candidates = [candidate for candidate in all_candidates
                                       if not candidate.version.is_prerelease]
 
                 evaluator = finder.make_candidate_evaluator(
-                    project_name=dist.project_name,
+                    project_name=dist.canonical_name,
                 )
                 best_candidate = evaluator.sort_best_candidate(all_candidates)
                 if best_candidate is None:
@@ -219,7 +232,6 @@ def latest_info(dist):
                     typ = 'wheel'
                 else:
                     typ = 'sdist'
-                # This is dirty but makes the rest of the code much cleaner
                 dist.latest_version = remote_version
                 dist.latest_filetype = typ
                 return dist
@@ -229,21 +241,21 @@ def latest_info(dist):
                     yield dist
 
     def output_package_listing(self, packages, options):
-        # type: (List[Distribution], Values) -> None
+        # type: (_ProcessedDists, Values) -> None
         packages = sorted(
             packages,
-            key=lambda dist: dist.project_name.lower(),
+            key=lambda dist: dist.canonical_name,
         )
         if options.list_format == 'columns' and packages:
             data, header = format_for_columns(packages, options)
             self.output_package_listing_columns(data, header)
         elif options.list_format == 'freeze':
             for dist in packages:
                 if options.verbose >= 1:
-                    write_output("%s==%s (%s)", dist.project_name,
+                    write_output("%s==%s (%s)", dist.canonical_name,
                                  dist.version, dist.location)
                 else:
-                    write_output("%s==%s", dist.project_name, dist.version)
+                    write_output("%s==%s", dist.canonical_name, dist.version)
         elif options.list_format == 'json':
             write_output(format_for_json(packages, options))
 
@@ -264,7 +276,7 @@ def output_package_listing_columns(self, data, header):
 
 
 def format_for_columns(pkgs, options):
-    # type: (List[Distribution], Values) -> Tuple[List[List[str]], List[str]]
+    # type: (_ProcessedDists, Values) -> Tuple[List[List[str]], List[str]]
     """
     Convert the package data into something usable
     by output_package_listing_columns.
@@ -277,41 +289,41 @@ def format_for_columns(pkgs, options):
         header = ["Package", "Version"]
 
     data = []
-    if options.verbose >= 1 or any(dist_is_editable(x) for x in pkgs):
+    if options.verbose >= 1 or any(x.editable for x in pkgs):
         header.append("Location")
     if options.verbose >= 1:
         header.append("Installer")
 
     for proj in pkgs:
         # if we're working on the 'outdated' list, separate out the
         # latest_version and type
-        row = [proj.project_name, proj.version]
+        row = [proj.canonical_name, str(proj.version)]
 
         if running_outdated:
-            row.append(proj.latest_version)
+            row.append(str(proj.latest_version))
             row.append(proj.latest_filetype)
 
-        if options.verbose >= 1 or dist_is_editable(proj):
-            row.append(proj.location)
+        if options.verbose >= 1 or proj.editable:
+            row.append(proj.location or "")
         if options.verbose >= 1:
-            row.append(get_installer(proj))
+            row.append(proj.installer)
 
         data.append(row)
 
     return data, header
 
 
 def format_for_json(packages, options):
-    # type: (List[Distribution], Values) -> str
+    # type: (_ProcessedDists, Values) -> str
     data = []
     for dist in packages:
         info = {
-            'name': dist.project_name,
+            'name': dist.canonical_name,
             'version': str(dist.version),
         }
         if options.verbose >= 1:
-            info['location'] = dist.location
-            info['installer'] = get_installer(dist)
+            info['location'] = dist.location or ""
+            info['installer'] = dist.installer
         if options.outdated:
             info['latest_version'] = str(dist.latest_version)
             info['latest_filetype'] = dist.latest_filetype

src/pip/_internal/metadata/base.py

@@ -1,17 +1,32 @@
 import logging
 import re
-from typing import Container, Iterator, List, Optional, Union
-
+from typing import (
+    TYPE_CHECKING,
+    Collection,
+    Container,
+    Iterable,
+    Iterator,
+    List,
+    Optional,
+    Union,
+)
+
+from pip._vendor.packaging.requirements import Requirement
 from pip._vendor.packaging.version import LegacyVersion, Version
 
 from pip._internal.utils.misc import stdlib_pkgs  # TODO: Move definition here.
 
+if TYPE_CHECKING:
+    from typing import Protocol
+else:
+    Protocol = object
+
 DistributionVersion = Union[LegacyVersion, Version]
 
 logger = logging.getLogger(__name__)
 
 
-class BaseDistribution:
+class BaseDistribution(Protocol):
     @property
     def location(self) -> Optional[str]:
         """Where the distribution is loaded from.
@@ -51,6 +66,10 @@ def local(self) -> bool:
     def in_usersite(self) -> bool:
         raise NotImplementedError()
 
+    def iter_dependencies(self, extras=()):
+        # type: (Collection[str]) -> Iterable[Requirement]
+        raise NotImplementedError()
+
 
 class BaseEnvironment:
     """An environment containing distributions to introspect."""

src/pip/_internal/metadata/pkg_resources.py

@@ -1,7 +1,9 @@
+import logging
 import zipfile
-from typing import Iterator, List, Optional
+from typing import Collection, Iterable, Iterator, List, Optional
 
 from pip._vendor import pkg_resources
+from pip._vendor.packaging.requirements import Requirement
 from pip._vendor.packaging.utils import canonicalize_name
 from pip._vendor.packaging.version import parse as parse_version
 
@@ -11,6 +13,8 @@
 
 from .base import BaseDistribution, BaseEnvironment, DistributionVersion
 
+logger = logging.getLogger(__name__)
+
 
 class Distribution(BaseDistribution):
     def __init__(self, dist: pkg_resources.Distribution) -> None:
@@ -57,6 +61,19 @@ def local(self) -> bool:
     def in_usersite(self) -> bool:
         return misc.dist_in_usersite(self._dist)
 
+    def iter_dependencies(self, extras=()):
+        # type: (Collection[str]) -> Iterable[Requirement]
+        # pkg_resources raises on invalid extras, so we sanitize.
+        requested_extras = set(extras)
+        valid_extras = requested_extras & set(self._dist.extras)
+        for invalid_extra in requested_extras ^ valid_extras:
+            logger.warning(
+                "Invalid extra %r for package %r discarded",
+                invalid_extra,
+                self.canonical_name,
+            )
+        return self._dist.requires(valid_extras)
+
 
 class Environment(BaseEnvironment):
     def __init__(self, ws: pkg_resources.WorkingSet) -> None:

tests/functional/test_list.py

@@ -100,10 +100,10 @@ def test_multiple_exclude_and_normalization(script, tmpdir):
     script.pip("install", "--no-index", req_path)
     result = script.pip("list")
     print(result.stdout)
-    assert "Normalizable-Name" in result.stdout
+    assert "normalizable-name" in result.stdout
     assert "pip" in result.stdout
     result = script.pip("list", "--exclude", "normalizablE-namE", "--exclude", "pIp")
-    assert "Normalizable-Name" not in result.stdout
+    assert "normalizable-name" not in result.stdout
     assert "pip" not in result.stdout
 
 
@@ -477,10 +477,10 @@ def test_not_required_flag(script, data):
         'install', '-f', data.find_links, '--no-index', 'TopoRequires4'
     )
     result = script.pip('list', '--not-required', expect_stderr=True)
-    assert 'TopoRequires4 ' in result.stdout, str(result)
-    assert 'TopoRequires ' not in result.stdout
-    assert 'TopoRequires2 ' not in result.stdout
-    assert 'TopoRequires3 ' not in result.stdout
+    assert 'toporequires4 ' in result.stdout, str(result)
+    assert 'toporequires ' not in result.stdout
+    assert 'toporequires2 ' not in result.stdout
+    assert 'toporequires3 ' not in result.stdout
 
 
 def test_list_freeze(simple_script):

tests/functional/test_new_resolver.py

@@ -1,10 +1,8 @@
-import json
 import os
 import sys
 import textwrap
 
 import pytest
-from pip._vendor.packaging.utils import canonicalize_name
 
 from tests.lib import (
     create_basic_sdist_for_package,
@@ -17,26 +15,14 @@
 from tests.lib.wheel import make_wheel
 
 
+# TODO: Remove me.
 def assert_installed(script, **kwargs):
-    ret = script.pip('list', '--format=json')
-    installed = {
-        (canonicalize_name(val['name']), val['version'])
-        for val in json.loads(ret.stdout)
-    }
-    expected = {(canonicalize_name(k), v) for k, v in kwargs.items()}
-    assert expected <= installed, f"{expected!r} not all in {installed!r}"
+    script.assert_installed(**kwargs)
 
 
+# TODO: Remove me.
 def assert_not_installed(script, *args):
-    ret = script.pip("list", "--format=json")
-    installed = {
-        canonicalize_name(val["name"])
-        for val in json.loads(ret.stdout)
-    }
-    # None of the given names should be listed as installed, i.e. their
-    # intersection should be empty.
-    expected = {canonicalize_name(k) for k in args}
-    assert not (expected & installed), f"{expected!r} contained in {installed!r}"
+    script.assert_not_installed(*args)
 
 
 def assert_editable(script, *args):