implement --report

- create LinkWithSource to retain attrs from InstallRequirement
- add tests for report output for top-level requirements
- add tests for more of the report JSON format
- add passing tests for JSON report output including PEP 658!
- add docstrings to several classes and functions, including tests!
- move the --report implementation into resolvelib
- use an abstract base class instead of a Union for InfoType
- use frozen dataclasses for InfoType subclasses

Author: cosmicexplorer

src/pip/_internal/commands/download.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import os
 from optparse import Values
@@ -7,7 +8,10 @@
 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.req.req_tracker import get_requirement_tracker
+from pip._internal.resolution.base import RequirementSetWithCandidates
+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
 
@@ -66,7 +70,22 @@ def add_options(self) -> None:
             "--dry-run",
             dest="dry_run",
             action="store_true",
-            help="Avoid actually downloading wheels.",
+            help=(
+                "Avoid actually downloading wheels or sdists. "
+                "Intended to be used with --report."
+            ),
+        )
+
+        self.cmd_opts.add_option(
+            "--report",
+            "--resolution-report",
+            dest="json_report_file",
+            metavar="file",
+            default=None,
+            help=(
+                "Print a JSON object representing the resolve into <file>. "
+                "Often used with --dry-run."
+            ),
         )
 
         cmdoptions.add_target_python_options(self.cmd_opts)
@@ -145,4 +164,25 @@ 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):
+            raise CommandError(
+                "The legacy resolver is being used via "
+                "--use-deprecated=legacy-resolver."
+                "The legacy resolver does not retain detailed dependency information, "
+                "so `pip download --report` cannot be used with it. "
+            )
+
+        resolution_result = ResolutionResult.generate_resolve_report(
+            reqs, requirement_set
+        )
+
+        # Write the full report data to the JSON output file.
+        with open(options.json_report_file, "w") as f:
+            json.dump(resolution_result.to_dict(), f, indent=4)
+        write_output(f"JSON report written to '{options.json_report_file}'.")
+
         return SUCCESS

src/pip/_internal/index/collector.py

@@ -8,10 +8,8 @@
 import itertools
 import logging
 import os
-import re
 import urllib.parse
 import urllib.request
-import xml.etree.ElementTree
 from optparse import Values
 from typing import (
     Callable,
@@ -29,19 +27,18 @@
 from pip._vendor.requests.exceptions import RetryError, SSLError
 
 from pip._internal.exceptions import NetworkConnectionError
-from pip._internal.models.link import Link
+from pip._internal.models.link import HTMLElement, Link
 from pip._internal.models.search_scope import SearchScope
 from pip._internal.network.session import PipSession
 from pip._internal.network.utils import raise_for_status
 from pip._internal.utils.filetypes import is_archive_file
-from pip._internal.utils.misc import pairwise, redact_auth_from_url
+from pip._internal.utils.misc import redact_auth_from_url
 from pip._internal.vcs import vcs
 
 from .sources import CandidatesFromPage, LinkSource, build_source
 
 logger = logging.getLogger(__name__)
 
-HTMLElement = xml.etree.ElementTree.Element
 ResponseHeaders = MutableMapping[str, str]
 
 
@@ -171,94 +168,6 @@ def _determine_base_url(document: HTMLElement, page_url: str) -> str:
     return page_url
 
 
-def _clean_url_path_part(part: str) -> str:
-    """
-    Clean a "part" of a URL path (i.e. after splitting on "@" characters).
-    """
-    # We unquote prior to quoting to make sure nothing is double quoted.
-    return urllib.parse.quote(urllib.parse.unquote(part))
-
-
-def _clean_file_url_path(part: str) -> str:
-    """
-    Clean the first part of a URL path that corresponds to a local
-    filesystem path (i.e. the first part after splitting on "@" characters).
-    """
-    # We unquote prior to quoting to make sure nothing is double quoted.
-    # Also, on Windows the path part might contain a drive letter which
-    # should not be quoted. On Linux where drive letters do not
-    # exist, the colon should be quoted. We rely on urllib.request
-    # to do the right thing here.
-    return urllib.request.pathname2url(urllib.request.url2pathname(part))
-
-
-# percent-encoded:                   /
-_reserved_chars_re = re.compile("(@|%2F)", re.IGNORECASE)
-
-
-def _clean_url_path(path: str, is_local_path: bool) -> str:
-    """
-    Clean the path portion of a URL.
-    """
-    if is_local_path:
-        clean_func = _clean_file_url_path
-    else:
-        clean_func = _clean_url_path_part
-
-    # Split on the reserved characters prior to cleaning so that
-    # revision strings in VCS URLs are properly preserved.
-    parts = _reserved_chars_re.split(path)
-
-    cleaned_parts = []
-    for to_clean, reserved in pairwise(itertools.chain(parts, [""])):
-        cleaned_parts.append(clean_func(to_clean))
-        # Normalize %xx escapes (e.g. %2f -> %2F)
-        cleaned_parts.append(reserved.upper())
-
-    return "".join(cleaned_parts)
-
-
-def _clean_link(url: str) -> str:
-    """
-    Make sure a link is fully quoted.
-    For example, if ' ' occurs in the URL, it will be replaced with "%20",
-    and without double-quoting other characters.
-    """
-    # Split the URL into parts according to the general structure
-    # `scheme://netloc/path;parameters?query#fragment`.
-    result = urllib.parse.urlparse(url)
-    # If the netloc is empty, then the URL refers to a local filesystem path.
-    is_local_path = not result.netloc
-    path = _clean_url_path(result.path, is_local_path=is_local_path)
-    return urllib.parse.urlunparse(result._replace(path=path))
-
-
-def _create_link_from_element(
-    anchor: HTMLElement,
-    page_url: str,
-    base_url: str,
-) -> Optional[Link]:
-    """
-    Convert an anchor element in a simple repository page to a Link.
-    """
-    href = anchor.get("href")
-    if not href:
-        return None
-
-    url = _clean_link(urllib.parse.urljoin(base_url, href))
-    pyrequire = anchor.get("data-requires-python")
-    yanked_reason = anchor.get("data-yanked")
-
-    link = Link(
-        url,
-        comes_from=page_url,
-        requires_python=pyrequire,
-        yanked_reason=yanked_reason,
-    )
-
-    return link
-
-
 class CacheablePageContent:
     def __init__(self, page: "HTMLPage") -> None:
         assert page.cache_link_parsing
@@ -307,11 +216,7 @@ def parse_links(page: "HTMLPage") -> Iterable[Link]:
     url = page.url
     base_url = _determine_base_url(document, url)
     for anchor in document.findall(".//a"):
-        link = _create_link_from_element(
-            anchor,
-            page_url=url,
-            base_url=base_url,
-        )
+        link = Link.from_element(anchor, page_url=url, base_url=base_url)
         if link is None:
             continue
         yield link

src/pip/_internal/metadata/base.py

@@ -101,6 +101,9 @@ def __repr__(self) -> str:
     def __str__(self) -> str:
         return f"{self.raw_name} {self.version}"
 
+    def as_serializable_requirement(self) -> Requirement:
+        raise NotImplementedError()
+
     @property
     def location(self) -> Optional[str]:
         """Where the distribution is loaded from.

src/pip/_internal/metadata/pkg_resources.py

@@ -120,6 +120,9 @@ def from_wheel(cls, wheel: Wheel, name: str) -> "Distribution":
         )
         return cls(dist)
 
+    def as_serializable_requirement(self) -> Requirement:
+        return self._dist.as_requirement()
+
     @property
     def location(self) -> Optional[str]:
         return self._dist.location

src/pip/_internal/models/direct_url.py

@@ -1,8 +1,10 @@
 """ PEP 610 """
+import abc
 import json
 import re
 import urllib.parse
-from typing import Any, Dict, Iterable, Optional, Type, TypeVar, Union
+from dataclasses import dataclass
+from typing import Any, ClassVar, Dict, Iterable, Optional, Type, TypeVar
 
 __all__ = [
     "DirectUrl",
@@ -47,8 +49,39 @@ def _get_required(
     return value
 
 
-def _exactly_one_of(infos: Iterable[Optional["InfoType"]]) -> "InfoType":
-    infos = [info for info in infos if info is not None]
+def _filter_none(**kwargs: Any) -> Dict[str, Any]:
+    """Make dict excluding None values."""
+    return {k: v for k, v in kwargs.items() if v is not None}
+
+
+class InfoType(metaclass=abc.ABCMeta):
+    """Superclass for the types of metadata that can be stored within a "direct URL"."""
+
+    name: ClassVar[str]
+
+    @classmethod
+    @abc.abstractmethod
+    def _from_dict(cls: Type[T], d: Optional[Dict[str, Any]]) -> Optional[T]:
+        """Parse an instance of this class from a JSON-serializable dict."""
+
+    @abc.abstractmethod
+    def _to_dict(self) -> Dict[str, Any]:
+        """Produce a JSON-serializable dict which can be parsed with `._from_dict()`."""
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "InfoType":
+        """Parse exactly one of the known subclasses from the dict `d`."""
+        return _exactly_one_of(
+            [
+                ArchiveInfo._from_dict(_get(d, dict, "archive_info")),
+                DirInfo._from_dict(_get(d, dict, "dir_info")),
+                VcsInfo._from_dict(_get(d, dict, "vcs_info")),
+            ]
+        )
+
+
+def _exactly_one_of(infos: Iterable[Optional[InfoType]]) -> InfoType:
+    infos = list(filter(None, infos))
     if not infos:
         raise DirectUrlValidationError(
             "missing one of archive_info, dir_info, vcs_info"
@@ -61,27 +94,15 @@ def _exactly_one_of(infos: Iterable[Optional["InfoType"]]) -> "InfoType":
     return infos[0]
 
 
-def _filter_none(**kwargs: Any) -> Dict[str, Any]:
-    """Make dict excluding None values."""
-    return {k: v for k, v in kwargs.items() if v is not None}
-
-
-class VcsInfo:
-    name = "vcs_info"
+@dataclass(frozen=True)
+class VcsInfo(InfoType):
+    vcs: str
+    commit_id: str
+    requested_revision: Optional[str] = None
+    resolved_revision: Optional[str] = None
+    resolved_revision_type: Optional[str] = None
 
-    def __init__(
-        self,
-        vcs: str,
-        commit_id: str,
-        requested_revision: Optional[str] = None,
-        resolved_revision: Optional[str] = None,
-        resolved_revision_type: Optional[str] = None,
-    ) -> None:
-        self.vcs = vcs
-        self.requested_revision = requested_revision
-        self.commit_id = commit_id
-        self.resolved_revision = resolved_revision
-        self.resolved_revision_type = resolved_revision_type
+    name: ClassVar[str] = "vcs_info"
 
     @classmethod
     def _from_dict(cls, d: Optional[Dict[str, Any]]) -> Optional["VcsInfo"]:
@@ -105,14 +126,11 @@ def _to_dict(self) -> Dict[str, Any]:
         )
 
 
-class ArchiveInfo:
-    name = "archive_info"
+@dataclass(frozen=True)
+class ArchiveInfo(InfoType):
+    hash: Optional[str] = None
 
-    def __init__(
-        self,
-        hash: Optional[str] = None,
-    ) -> None:
-        self.hash = hash
+    name: ClassVar[str] = "archive_info"
 
     @classmethod
     def _from_dict(cls, d: Optional[Dict[str, Any]]) -> Optional["ArchiveInfo"]:
@@ -124,14 +142,11 @@ def _to_dict(self) -> Dict[str, Any]:
         return _filter_none(hash=self.hash)
 
 
-class DirInfo:
-    name = "dir_info"
+@dataclass(frozen=True)
+class DirInfo(InfoType):
+    editable: bool = False
 
-    def __init__(
-        self,
-        editable: bool = False,
-    ) -> None:
-        self.editable = editable
+    name: ClassVar[str] = "dir_info"
 
     @classmethod
     def _from_dict(cls, d: Optional[Dict[str, Any]]) -> Optional["DirInfo"]:
@@ -143,19 +158,11 @@ def _to_dict(self) -> Dict[str, Any]:
         return _filter_none(editable=self.editable or None)
 
 
-InfoType = Union[ArchiveInfo, DirInfo, VcsInfo]
-
-
+@dataclass(frozen=True)
 class DirectUrl:
-    def __init__(
-        self,
-        url: str,
-        info: InfoType,
-        subdirectory: Optional[str] = None,
-    ) -> None:
-        self.url = url
-        self.info = info
-        self.subdirectory = subdirectory
+    url: str
+    info: InfoType
+    subdirectory: Optional[str] = None
 
     def _remove_auth_from_netloc(self, netloc: str) -> str:
         if "@" not in netloc:
@@ -192,13 +199,7 @@ def from_dict(cls, d: Dict[str, Any]) -> "DirectUrl":
         return DirectUrl(
             url=_get_required(d, str, "url"),
             subdirectory=_get(d, str, "subdirectory"),
-            info=_exactly_one_of(
-                [
-                    ArchiveInfo._from_dict(_get(d, dict, "archive_info")),
-                    DirInfo._from_dict(_get(d, dict, "dir_info")),
-                    VcsInfo._from_dict(_get(d, dict, "vcs_info")),
-                ]
-            ),
+            info=InfoType.from_dict(d),
         )
 
     def to_dict(self) -> Dict[str, Any]: