Add pip lock command (PEP 751)

Author: sbidoul

src/pip/_internal/commands/__init__.py

@@ -23,6 +23,11 @@
         "InstallCommand",
         "Install packages.",
     ),
+    "lock": CommandInfo(
+        "pip._internal.commands.lock",
+        "LockCommand",
+        "Generate a lock file.",
+    ),
     "download": CommandInfo(
         "pip._internal.commands.download",
         "DownloadCommand",

src/pip/_internal/commands/lock.py

@@ -0,0 +1,138 @@
+import sys
+from optparse import Values
+from typing import List
+
+from pip._internal.cache import WheelCache
+from pip._internal.cli import cmdoptions
+from pip._internal.cli.req_command import (
+    RequirementCommand,
+    with_cleanup,
+)
+from pip._internal.cli.status_codes import SUCCESS
+from pip._internal.models.pylock import Pylock
+from pip._internal.operations.build.build_tracker import get_build_tracker
+from pip._internal.req.req_install import (
+    check_legacy_setup_py_options,
+)
+from pip._internal.utils.logging import getLogger
+from pip._internal.utils.misc import (
+    get_pip_version,
+)
+from pip._internal.utils.temp_dir import TempDirectory
+
+logger = getLogger(__name__)
+
+
+class LockCommand(RequirementCommand):
+    """
+    Lock packages from:
+
+    - PyPI (and other indexes) using requirement specifiers.
+    - VCS project urls.
+    - Local project directories.
+    - Local or remote source archives.
+
+    pip also supports locking from "requirements files", which provide
+    an easy way to specify a whole environment to be installed.
+    """
+
+    usage = """
+      %prog [options] <requirement specifier> [package-index-options] ...
+      %prog [options] -r <requirements file> [package-index-options] ...
+      %prog [options] [-e] <vcs project url> ...
+      %prog [options] [-e] <local project path> ...
+      %prog [options] <archive url/path> ..."""
+
+    def add_options(self) -> None:
+        self.cmd_opts.add_option(cmdoptions.requirements())
+        self.cmd_opts.add_option(cmdoptions.constraints())
+        self.cmd_opts.add_option(cmdoptions.no_deps())
+        self.cmd_opts.add_option(cmdoptions.pre())
+
+        self.cmd_opts.add_option(cmdoptions.editable())
+
+        self.cmd_opts.add_option(cmdoptions.src())
+
+        self.cmd_opts.add_option(cmdoptions.ignore_requires_python())
+        self.cmd_opts.add_option(cmdoptions.no_build_isolation())
+        self.cmd_opts.add_option(cmdoptions.use_pep517())
+        self.cmd_opts.add_option(cmdoptions.no_use_pep517())
+        self.cmd_opts.add_option(cmdoptions.check_build_deps())
+
+        self.cmd_opts.add_option(cmdoptions.config_settings())
+
+        self.cmd_opts.add_option(cmdoptions.no_binary())
+        self.cmd_opts.add_option(cmdoptions.only_binary())
+        self.cmd_opts.add_option(cmdoptions.prefer_binary())
+        self.cmd_opts.add_option(cmdoptions.require_hashes())
+        self.cmd_opts.add_option(cmdoptions.progress_bar())
+
+        index_opts = cmdoptions.make_option_group(
+            cmdoptions.index_group,
+            self.parser,
+        )
+
+        self.parser.insert_option_group(0, index_opts)
+        self.parser.insert_option_group(0, self.cmd_opts)
+
+    @with_cleanup
+    def run(self, options: Values, args: List[str]) -> int:
+        logger.verbose("Using %s", get_pip_version())
+
+        session = self.get_default_session(options)
+
+        finder = self._build_package_finder(
+            options=options,
+            session=session,
+            ignore_requires_python=options.ignore_requires_python,
+        )
+        build_tracker = self.enter_context(get_build_tracker())
+
+        directory = TempDirectory(
+            delete=not options.no_clean,
+            kind="install",
+            globally_managed=True,
+        )
+
+        reqs = self.get_requirements(args, options, finder, session)
+        check_legacy_setup_py_options(options, reqs)
+
+        wheel_cache = WheelCache(options.cache_dir)
+
+        # Only when installing is it permitted to use PEP 660.
+        # In other circumstances (pip wheel, pip download) we generate
+        # regular (i.e. non editable) metadata and wheels.
+        for req in reqs:
+            req.permit_editable_wheels = True
+
+        preparer = self.make_requirement_preparer(
+            temp_build_dir=directory,
+            options=options,
+            build_tracker=build_tracker,
+            session=session,
+            finder=finder,
+            use_user_site=False,
+            verbosity=self.verbosity,
+        )
+        resolver = self.make_resolver(
+            preparer=preparer,
+            finder=finder,
+            options=options,
+            wheel_cache=wheel_cache,
+            use_user_site=False,
+            ignore_installed=True,
+            ignore_requires_python=options.ignore_requires_python,
+            upgrade_strategy="to-satisfy-only",
+            use_pep517=options.use_pep517,
+        )
+
+        self.trace_basic_info(finder)
+
+        requirement_set = resolver.resolve(reqs, check_supported_wheels=True)
+
+        pyproject_lock = Pylock.from_install_requirements(
+            requirement_set.requirements.values()
+        )
+        sys.stdout.write(pyproject_lock.as_toml())
+
+        return SUCCESS

src/pip/_internal/models/pylock.py

@@ -0,0 +1,170 @@
+import dataclasses
+from dataclasses import dataclass
+from typing import Any, Dict, Iterable, List, Optional, Tuple
+
+from pip._vendor import tomli_w
+from pip._vendor.typing_extensions import Self
+
+from pip._internal.models.direct_url import ArchiveInfo, DirInfo, VcsInfo
+from pip._internal.models.link import Link
+from pip._internal.req.req_install import InstallRequirement
+from pip._internal.utils.urls import url_to_path
+
+
+def _toml_dict_factory(data: List[Tuple[str, Any]]) -> Dict[str, Any]:
+    return {key.replace("_", "-"): value for key, value in data if value is not None}
+
+
+@dataclass
+class PackageVcs:
+    type: str
+    url: Optional[str]
+    # (not supported) path: Optional[str]
+    requested_revision: Optional[str]
+    commit_id: str
+    subdirectory: Optional[str]
+
+
+@dataclass
+class PackageDirectory:
+    path: str
+    editable: Optional[bool]
+    subdirectory: Optional[str]
+
+
+@dataclass
+class PackageArchive:
+    url: Optional[str]
+    # (not supported) path: Optional[str]
+    # (not supported) size: Optional[int]
+    hashes: Dict[str, str]
+    subdirectory: Optional[str]
+
+
+@dataclass
+class PackageSdist:
+    name: str
+    # (not supported) upload_time: Optional[datetime]
+    url: Optional[str]
+    # (not supported) path: Optional[str]
+    # (not supported) size: Optional[int]
+    hashes: Dict[str, str]
+
+
+@dataclass
+class PackageWheel:
+    name: str
+    # (not supported) upload_time: Optional[datetime]
+    url: Optional[str]
+    # (not supported) path: Optional[str]
+    # (not supported) size: Optional[int]
+    hashes: Dict[str, str]
+
+
+@dataclass
+class Package:
+    name: str
+    version: Optional[str] = None
+    # (not supported) marker: Optional[str]
+    # (not supported) requires_python: Optional[str]
+    # (not supported) dependencies
+    direct: Optional[bool] = None
+    vcs: Optional[PackageVcs] = None
+    directory: Optional[PackageDirectory] = None
+    archive: Optional[PackageArchive] = None
+    # (not supported) index: Optional[str]
+    sdist: Optional[PackageSdist] = None
+    wheels: Optional[List[PackageWheel]] = None
+    # (not supported) attestation_identities: Optional[List[Dict[str, Any]]]
+    # (not supported) tool: Optional[Dict[str, Any]]
+
+    @classmethod
+    def from_install_requirement(cls, ireq: InstallRequirement) -> Self:
+        dist = ireq.get_dist()
+        download_info = ireq.download_info
+        assert download_info
+        package = cls(
+            name=dist.canonical_name,
+            version=str(dist.version),
+        )
+        package.direct = ireq.is_direct if ireq.is_direct else None
+        if package.direct:
+            if isinstance(download_info.info, VcsInfo):
+                package.vcs = PackageVcs(
+                    type=download_info.info.vcs,
+                    url=download_info.url,
+                    requested_revision=download_info.info.requested_revision,
+                    commit_id=download_info.info.commit_id,
+                    subdirectory=download_info.subdirectory,
+                )
+            elif isinstance(download_info.info, DirInfo):
+                package.directory = PackageDirectory(
+                    path=url_to_path(download_info.url),
+                    editable=(
+                        download_info.info.editable
+                        if download_info.info.editable
+                        else None
+                    ),
+                    subdirectory=download_info.subdirectory,
+                )
+            elif isinstance(download_info.info, ArchiveInfo):
+                if not download_info.info.hashes:
+                    raise NotImplementedError()
+                package.archive = PackageArchive(
+                    url=download_info.url,
+                    hashes=download_info.info.hashes,
+                    subdirectory=download_info.subdirectory,
+                )
+            else:
+                # should never happen
+                raise NotImplementedError()
+        else:
+            if isinstance(download_info.info, ArchiveInfo):
+                if not download_info.info.hashes:
+                    raise NotImplementedError()
+                link = Link(download_info.url)
+                if link.is_wheel:
+                    package.wheels = [
+                        PackageWheel(
+                            name=link.filename,
+                            url=download_info.url,
+                            hashes=download_info.info.hashes,
+                        )
+                    ]
+                else:
+                    package.sdist = PackageSdist(
+                        name=link.filename,
+                        url=download_info.url,
+                        hashes=download_info.info.hashes,
+                    )
+            else:
+                # should never happen
+                raise NotImplementedError()
+        return package
+
+
+@dataclass
+class Pylock:
+    lock_version: str = "1.0"
+    # (not supported) environments: Optional[List[str]]
+    # (not supported) requires_python: Optional[str]
+    created_by: str = "pip"
+    packages: List[Package] = dataclasses.field(default_factory=list)
+    # (not supported) tool: Optional[Dict[str, Any]]
+
+    def as_toml(self) -> str:
+        return tomli_w.dumps(dataclasses.asdict(self, dict_factory=_toml_dict_factory))
+
+    @classmethod
+    def from_install_requirements(
+        cls, install_requirements: Iterable[InstallRequirement]
+    ) -> Self:
+        return cls(
+            packages=sorted(
+                (
+                    Package.from_install_requirement(ireq)
+                    for ireq in install_requirements
+                ),
+                key=lambda p: p.name,
+            )
+        )

tests/functional/test_cli.py

@@ -57,7 +57,7 @@ def test_entrypoints_work(entrypoint: str, script: PipTestEnvironment) -> None:
     sorted(
         set(commands_dict).symmetric_difference(
             # Exclude commands that are expected to use the network.
-            {"install", "download", "search", "index", "wheel"}
+            {"install", "download", "search", "index", "lock", "wheel"}
         )
     ),
 )

tests/unit/test_commands.py

@@ -14,7 +14,14 @@
 
 # These are the expected names of the commands whose classes inherit from
 # IndexGroupCommand.
-EXPECTED_INDEX_GROUP_COMMANDS = ["download", "index", "install", "list", "wheel"]
+EXPECTED_INDEX_GROUP_COMMANDS = [
+    "download",
+    "index",
+    "install",
+    "list",
+    "lock",
+    "wheel",
+]
 
 
 def check_commands(pred: Callable[[Command], bool], expected: List[str]) -> None:
@@ -53,7 +60,16 @@ def test_session_commands() -> None:
     def is_session_command(command: Command) -> bool:
         return isinstance(command, SessionCommandMixin)
 
-    expected = ["download", "index", "install", "list", "search", "uninstall", "wheel"]
+    expected = [
+        "download",
+        "index",
+        "install",
+        "list",
+        "lock",
+        "search",
+        "uninstall",
+        "wheel",
+    ]
     check_commands(is_session_command, expected)
 
 
@@ -124,7 +140,7 @@ def test_requirement_commands() -> None:
     def is_requirement_command(command: Command) -> bool:
         return isinstance(command, RequirementCommand)
 
-    check_commands(is_requirement_command, ["download", "install", "wheel"])
+    check_commands(is_requirement_command, ["download", "install", "lock", "wheel"])
 
 
 @pytest.mark.parametrize("flag", ["", "--outdated", "--uptodate"])