New implementation using temporary lockfiles

poetry_monoranger_plugin/export_modifier.py

@@ -7,12 +7,14 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+import shutil
+import tempfile
+import weakref
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, TypeVar
 
 from poetry.config.config import Config
 from poetry.core.packages.dependency_group import MAIN_GROUP
-from poetry.core.packages.directory_dependency import DirectoryDependency
-from poetry.core.packages.project_package import ProjectPackage
 from poetry.factory import Factory
 from poetry.packages.locker import Locker
 from poetry.poetry import Poetry
@@ -23,69 +25,49 @@
 if TYPE_CHECKING:
     from cleo.events.console_command_event import ConsoleCommandEvent
     from cleo.io.io import IO
-    from poetry.core.packages.dependency import Dependency
     from poetry.core.packages.package import Package
 
     from poetry_monoranger_plugin.config import MonorangerConfig
 
+TemporaryLockerT = TypeVar("TemporaryLockerT", bound="TemporaryLocker")
 
-class PathDepPinningLocker(Locker):
-    """A modified Locker that pins path dependencies in the lock file to specific versions"""
 
-    _pinner: PathDepPinner
-    _io: IO
+class TemporaryLocker(Locker):
+    """A temporary locker that is used to store the lock file in a temporary file."""
 
     @classmethod
-    def from_locker(cls, locker: Locker, pinner: PathDepPinner, io: IO) -> PathDepPinningLocker:
-        """Creates a new PathDepPinningLocker from an existing Locker"""
-        new_locker = cls.__new__(cls)
-        new_locker.__dict__.update(locker.__dict__)
+    def from_locker(
+        cls: type[TemporaryLockerT], locker: Locker, pyproject_data: dict[str, Any] | None
+    ) -> TemporaryLockerT:
+        """Creates a temporary locker from an existing locker."""
+        temp_file = tempfile.NamedTemporaryFile(prefix="poetry_lock_", delete=False)  # noqa: SIM115
+        temp_file_path = Path(temp_file.name)
+        temp_file.close()
 
-        new_locker._io = io
-        new_locker._pinner = pinner
-        return new_locker
-
-    def _get_locked_package(self, info: dict[str, Any], with_dependencies: bool = True) -> Package:
-        package = super()._get_locked_package(info, with_dependencies)
-
-        if package.source_type == "directory":
-            package._source_type = None
-            package._source_url = None
+        shutil.copy(locker.lock, temp_file_path)
 
-            # noinspection PyProtectedMember
-            if package._dependency_groups and MAIN_GROUP in package._dependency_groups:
-                # noinspection PyProtectedMember
-                main_deps_group = package._dependency_groups[MAIN_GROUP]
-                # noinspection PyProtectedMember
-                self._pinner._pin_dep_grp(main_deps_group, self._io)
+        if pyproject_data is None:
+            pyproject_data = locker._pyproject_data
 
-        return package
+        new_locker: TemporaryLockerT = cls(temp_file_path, pyproject_data)
+        weakref.finalize(new_locker, temp_file_path.unlink)
 
+        return new_locker
 
-class PathDepPinningPackage(ProjectPackage):
-    """A modified ProjectPackage that pins path dependencies to specific versions"""
 
-    _pinner: PathDepPinner
+def _pin_package(package: Package, pinner: PathDepPinner, io: IO) -> Package:
+    """Pins a package to a specific version if it is a path dependency"""
+    if package.source_type == "directory":
+        package._source_type = None
+        package._source_url = None
 
-    @classmethod
-    def from_package(cls, package: Package, pinner: PathDepPinner) -> PathDepPinningPackage:
-        """Creates a new PathDepPinningPackage from an existing Package"""
-        new_package = cls.__new__(cls)
-        new_package.__dict__.update(package.__dict__)
-
-        new_package._pinner = pinner
-        return new_package
-
-    @property
-    def all_requires(self) -> list[Dependency]:
-        """Returns the main dependencies and group dependencies
-        enriched with Poetry-specific information for locking while ensuring
-        path dependencies are pinned to specific versions.
-        """
-        deps = super().all_requires
         # noinspection PyProtectedMember
-        deps = [self._pinner._pin_dependency(dep) if isinstance(dep, DirectoryDependency) else dep for dep in deps]
-        return deps
+        if package._dependency_groups and MAIN_GROUP in package._dependency_groups:
+            # noinspection PyProtectedMember
+            main_deps_group = package._dependency_groups[MAIN_GROUP]
+            # noinspection PyProtectedMember
+            pinner._pin_dep_grp(main_deps_group, io)
+    return package
 
 
 class ExportModifier:
@@ -116,22 +98,38 @@ def execute(self, event: ConsoleCommandEvent):
         io = event.io
         io.write_line("<info>Running command from monorepo root directory</info>")
 
+        # Create a copy of the poetry object to prevent the command from modifying the original poetry object
+        poetry = Poetry.__new__(Poetry)
+        poetry.__dict__.update(command.poetry.__dict__)
+
         # Force reload global config in order to undo changes that happened due to subproject's poetry.toml configs
         _ = Config.create(reload=True)
         monorepo_root = (command.poetry.pyproject_path.parent / self.plugin_conf.monorepo_root).resolve()
         monorepo_root_poetry = Factory().create_poetry(
             cwd=monorepo_root, io=io, disable_cache=command.poetry.disable_cache
         )
 
-        pinner = PathDepPinner(self.plugin_conf)
+        temp_locker = TemporaryLocker.from_locker(monorepo_root_poetry.locker, poetry.pyproject.data)
 
-        # Create a copy of the poetry object to prevent the command from modifying the original poetry object
-        poetry = Poetry.__new__(Poetry)
-        poetry.__dict__.update(command.poetry.__dict__)
-        pinning_package = PathDepPinningPackage.from_package(poetry.package, pinner)
-        poetry._package = pinning_package
+        from poetry.puzzle.solver import Solver
+
+        locked_repository = monorepo_root_poetry.locker.locked_repository()
+        solver = Solver(
+            poetry.package,
+            poetry.pool,
+            locked_repository.packages,
+            locked_repository.packages,
+            io,
+        )
 
-        pinning_locker = PathDepPinningLocker.from_locker(monorepo_root_poetry.locker, pinner, io)
-        poetry.set_locker(pinning_locker)
+        # Always re-solve directory dependencies, otherwise we can't determine
+        # if anything has changed (and the lock file contains an invalid version).
+        use_latest = [p.name for p in locked_repository.packages if p.source_type == "directory"]
+        packages = solver.solve(use_latest=use_latest).get_solved_packages()
+
+        pinner = PathDepPinner(self.plugin_conf)
+        packages = {_pin_package(pak, pinner, io): info for pak, info in packages.items()}
 
+        temp_locker.set_lock_data(poetry.package, packages)
+        poetry.set_locker(temp_locker)
         command.set_poetry(poetry)

tests/test_commands.py

@@ -196,14 +196,27 @@ def test_build(repo_manager, poetry_run, repo_name):
     assert "Requires-Dist: pkg-one[withfsspec] (==0.1.0)" in pkg_info_path.read_text()
 
 
-# @pytest.mark.parametrize(
-#     "repo_name",
-#     [pytest.param("v1"), pytest.param("v1_v2lock", marks=only_poetry_v2), pytest.param("v2", marks=only_poetry_v2)],
-# )
-# def test_export(repo_manager, poetry_run, repo_name):
-#     # Arrange
-#     root_dir = repo_manager.get_repo(repo_name, preinstalled=False)
-#
-#     # Act
-#     result = poetry_run(root_dir, "pkg_three", "export --format requirements.txt")
-#     print("done")
+@pytest.mark.parametrize(
+    "repo_name",
+    [
+        pytest.param("v1"),
+        pytest.param("v1_v2lock", marks=only_poetry_v2),
+        pytest.param(
+            "v2", marks=[only_poetry_v2, pytest.mark.skip(reason="not working currently for PEP621 projects")]
+        ),
+    ],
+)
+def test_export(repo_manager, poetry_run, repo_name):
+    # Arrange
+    root_dir = repo_manager.get_repo(repo_name, preinstalled=False)
+
+    # Act
+    result = poetry_run(
+        root_dir, "pkg_three", "export --format requirements.txt --without-hashes --output requirements.txt"
+    )
+    requirements_txt = (root_dir / "pkg_three" / "requirements.txt").read_text()
+
+    assert result.exit_code == 0
+    assert len(requirements_txt.strip().split("\n")) == 2  # 2 packages in requirements.txt
+    assert "fsspec" in requirements_txt
+    assert "pkg-one" in requirements_txt