move the --report implementation into resolvelib

Author: cosmicexplorer

src/pip/_internal/commands/download.py

@@ -1,91 +1,23 @@
 import json
 import logging
 import os
-from dataclasses import dataclass, field
 from optparse import Values
-from typing import Any, Dict, List, Optional, Tuple
-
-from pip._vendor.packaging.requirements import Requirement
-from pip._vendor.packaging.specifiers import SpecifierSet
+from typing import List
 
 from pip._internal.cli import cmdoptions
 from pip._internal.cli.cmdoptions import make_target_python
 from pip._internal.cli.req_command import RequirementCommand, with_cleanup
 from pip._internal.cli.status_codes import SUCCESS
 from pip._internal.exceptions import CommandError
-from pip._internal.models.link import LinkWithSource, URLDownloadInfo
-from pip._internal.req.req_install import produce_exact_version_specifier
 from pip._internal.req.req_tracker import get_requirement_tracker
 from pip._internal.resolution.base import RequirementSetWithCandidates
-from pip._internal.resolution.resolvelib.candidates import (
-    LinkCandidate,
-    RequiresPythonCandidate,
-)
-from pip._internal.resolution.resolvelib.requirements import (
-    ExplicitRequirement,
-    RequiresPythonRequirement,
-)
+from pip._internal.resolution.resolvelib.reporter import ResolutionResult
 from pip._internal.utils.misc import ensure_dir, normalize_path, write_output
 from pip._internal.utils.temp_dir import TempDirectory
 
 logger = logging.getLogger(__name__)
 
 
-@dataclass(frozen=True)
-class ResolvedCandidate:
-    """Coalesce all the information pip's resolver retains about an
-    installation candidate."""
-
-    req: Requirement
-    download_info: URLDownloadInfo
-    dependencies: Tuple[Requirement, ...]
-    requires_python: Optional[SpecifierSet]
-
-    def as_json(self) -> Dict[str, Any]:
-        """Return a JSON-serializable representation of this install candidate."""
-        return {
-            "requirement": str(self.req),
-            "download_info": self.download_info.as_json(),
-            "dependencies": {dep.name: str(dep) for dep in self.dependencies},
-            "requires_python": str(self.requires_python)
-            if self.requires_python
-            else None,
-        }
-
-
-@dataclass
-class ResolutionResult:
-    """The inputs and outputs of a pip internal resolve process."""
-
-    input_requirements: Tuple[str, ...]
-    python_version: Optional[SpecifierSet] = None
-    candidates: Dict[str, ResolvedCandidate] = field(default_factory=dict)
-
-    def as_basic_log(self, output_json_path: str) -> str:
-        """Generate a summary of the detailed JSON report produced with --report."""
-        inputs = " ".join(f"'{req}'" for req in self.input_requirements)
-        resolved = " ".join(f"'{info.req}'" for info in self.candidates.values())
-        return "\n".join(
-            [
-                f"Python version: '{self.python_version}'",
-                f"Input requirements: {inputs}",
-                f"Resolution: {resolved}",
-                f"JSON report written to '{output_json_path}'.",
-            ]
-        )
-
-    def as_json(self) -> Dict[str, Any]:
-        """Return a JSON-serializable representation of the resolve process."""
-        return {
-            "experimental": True,
-            "input_requirements": [str(req) for req in self.input_requirements],
-            "python_version": str(self.python_version),
-            "candidates": {
-                name: info.as_json() for name, info in self.candidates.items()
-            },
-        }
-
-
 class DownloadCommand(RequirementCommand):
     """
     Download packages from:
@@ -220,13 +152,6 @@ def run(self, options: Values, args: List[str]) -> int:
 
         self.trace_basic_info(finder)
 
-        # TODO: for performance, try to decouple extracting sdist metadata from
-        # actually building the sdist. See https://github.com/pypa/pip/issues/8929.
-        # As mentioned in that issue, PEP 658 support on PyPI would address many cases,
-        # but it would drastically improve performance for many existing packages if we
-        # attempted to extract PKG-INFO or .egg-info from non-wheel files, falling back
-        # to the slower setup.py invocation if not found. LazyZipOverHTTP and
-        # MemoryWheel already implement such a hack for wheel files specifically.
         requirement_set = resolver.resolve(reqs, check_supported_wheels=True)
 
         if not options.dry_run:
@@ -239,6 +164,8 @@ def run(self, options: Values, args: List[str]) -> int:
             if downloaded:
                 write_output("Successfully downloaded %s", " ".join(downloaded))
 
+        # The rest of this method pertains to generating the ResolutionReport with
+        # --report.
         if not options.json_report_file:
             return SUCCESS
         if not isinstance(requirement_set, RequirementSetWithCandidates):
@@ -249,98 +176,13 @@ def run(self, options: Values, args: List[str]) -> int:
                 "so `pip download --report` cannot be used with it. "
             )
 
-        # Reconstruct the input requirements provided to the resolve.
-        input_requirements: List[str] = []
-        for ireq in reqs:
-            if ireq.req:
-                # If the initial requirement string contained a url (retained in
-                # InstallRequirement.link), add it back to the requirement string
-                # included in the JSON report.
-                if ireq.link:
-                    req_string = f"{ireq.req}@{ireq.link.url}"
-                else:
-                    req_string = str(ireq.req)
-            else:
-                assert ireq.link
-                req_string = ireq.link.url
-
-            input_requirements.append(req_string)
-
-        # Scan all the elements of the resulting `RequirementSet` and map it back to all
-        # the install candidates preserved by `RequirementSetWithCandidates`.
-        resolution_result = ResolutionResult(
-            input_requirements=tuple(input_requirements)
+        resolution_result = ResolutionResult.generate_resolve_report(
+            reqs, requirement_set
         )
-        for candidate in requirement_set.candidates.mapping.values():
-            # This will occur for the python version requirement, for example.
-            if candidate.name not in requirement_set.requirements:
-                if isinstance(candidate, RequiresPythonCandidate):
-                    assert resolution_result.python_version is None
-                    resolution_result.python_version = produce_exact_version_specifier(
-                        str(candidate.version)
-                    )
-                    continue
-                raise TypeError(
-                    f"unknown candidate not found in requirement set: {candidate}"
-                )
-
-            req = requirement_set.requirements[candidate.name]
-            assert req.name is not None
-            assert req.link is not None
-            assert req.name not in resolution_result.candidates
-
-            # Scan the dependencies of the installation candidates, which cover both
-            # normal dependencies as well as Requires-Python information.
-            requires_python: Optional[SpecifierSet] = None
-            dependencies: List[Requirement] = []
-            for maybe_dep in candidate.iter_dependencies(with_requires=True):
-                # It's unclear why `.iter_dependencies()` may occasionally yield `None`.
-                if maybe_dep is None:
-                    continue
-                # There will only ever be one of these for each candidate, if any. We
-                # extract the version specifier.
-                if isinstance(maybe_dep, RequiresPythonRequirement):
-                    requires_python = maybe_dep.specifier
-                    continue
-
-                # Convert the 2020 resolver-internal Requirement subclass instance into
-                # a `packaging.requirements.Requirement` instance.
-                maybe_req = maybe_dep.as_serializable_requirement()
-                if maybe_req is None:
-                    continue
-
-                # For `ExplicitRequirement`s only, we want to make sure we propagate any
-                # source URL into a dependency's `packaging.requirements.Requirement`
-                # instance.
-                if isinstance(maybe_dep, ExplicitRequirement):
-                    dep_candidate = maybe_dep.candidate
-                    if maybe_req.url is None and isinstance(
-                        dep_candidate, LinkCandidate
-                    ):
-                        assert dep_candidate.source_link is not None
-                        maybe_req = Requirement(
-                            f"{maybe_req}@{dep_candidate.source_link.url}"
-                        )
-
-                dependencies.append(maybe_req)
-
-            # Mutate the candidates dictionary to add this candidate after processing
-            # any dependencies and python version requirement.
-            resolution_result.candidates[req.name] = ResolvedCandidate(
-                req=candidate.as_serializable_requirement(),
-                download_info=URLDownloadInfo.from_link_with_source(
-                    LinkWithSource(
-                        req.link,
-                        source_dir=req.source_dir,
-                        link_is_in_wheel_cache=req.original_link_is_in_wheel_cache,
-                    )
-                ),
-                dependencies=tuple(dependencies),
-                requires_python=requires_python,
-            )
 
         # Write a simplified representation of the resolution to stdout.
         write_output(resolution_result.as_basic_log(options.json_report_file))
+        # Write the full report data to the JSON output file.
         with open(options.json_report_file, "w") as f:
             json.dump(resolution_result.as_json(), f, indent=4)