Implement explicit lockfiles integration

.gitignore

@@ -130,3 +130,6 @@ dmypy.json
 
 # pixi
 .pixi/
+
+# Used in debugging
+explicit.txt

conda_pypi/__init__.py

@@ -1,3 +1,5 @@
 """
 conda-pypi
 """
+
+__version__ = "0.1.0"

conda_pypi/cli/__init__.py

@@ -0,0 +1,3 @@
+from . import install, list, pip
+
+__all__ = ["install", "list", "pip"]

conda_pypi/cli/install.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import sys
+from logging import getLogger
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from conda.base.context import context
+from conda.common.io import Spinner
+from conda.exceptions import CondaVerificationError, CondaFileIOError
+
+from ..main import run_pip_install, compute_record_sum, PyPIDistribution
+from ..utils import get_env_site_packages
+
+if TYPE_CHECKING:
+    from typing import Iterable, Literal
+
+log = getLogger(f"conda.{__name__}")
+
+
+def _prepare_pypi_transaction(lines: Iterable[str]) -> dict[str, dict[str, str]]:
+    pkgs = {}
+    for line in lines:
+        dist = PyPIDistribution.from_lockfile_line(line)
+        pkgs[(dist.name, dist.version)] = {
+            "url": dist.find_wheel_url(),
+            "hashes": dist.record_checksums,
+        }
+    return pkgs
+
+
+def _verify_pypi_transaction(
+    prefix: str,
+    pkgs: dict[str, dict[str, str]],
+    on_error: Literal["ignore", "warn", "error"] = "warn",
+):
+    site_packages = get_env_site_packages(prefix)
+    errors = []
+    dist_infos = [path for path in site_packages.glob("*.dist-info") if path.is_dir()]
+    for (name, version), pkg in pkgs.items():
+        norm_name = name.lower().replace("-", "_").replace(".", "_")
+        dist_info = next(
+            (
+                d
+                for d in dist_infos
+                if d.stem.rsplit("-", 1) in ([name, version], [norm_name, version])
+            ),
+            None,
+        )
+        if not dist_info:
+            errors.append(f"Could not find installation for {name}=={version}")
+            continue
+
+        expected_hashes = pkg.get("hashes")
+        if expected_hashes:
+            found_hashes = compute_record_sum(dist_info / "RECORD", expected_hashes.keys())
+            log.info("Verifying %s==%s with %s", name, version, ", ".join(expected_hashes))
+            for algo, expected_hash in expected_hashes.items():
+                found_hash = found_hashes.get(algo)
+                if found_hash and expected_hash != found_hash:
+                    msg = (
+                        "%s checksum for %s==%s didn't match! Expected=%s, found=%s",
+                        algo,
+                        name,
+                        version,
+                        expected_hash,
+                        found_hash,
+                    )
+                    if on_error == "warn":
+                        log.warning(*msg)
+                    elif on_error == "error":
+                        errors.append(msg[0] % msg[1:])
+                    else:
+                        log.debug(*msg)
+    if errors:
+        errors = "\n- ".join(errors)
+        raise CondaVerificationError(f"PyPI packages checksum verification failed:\n- {errors}")
+
+
+def post_command(command: str) -> int:
+    if command not in ("install", "create"):
+        return 0
+
+    pypi_lines = _pypi_lines_from_paths()
+    if not pypi_lines:
+        return 0
+
+    with Spinner("\nPreparing PyPI transaction", enabled=not context.quiet, json=context.json):
+        pkgs = _prepare_pypi_transaction(pypi_lines)
+
+    with Spinner("Executing PyPI transaction", enabled=not context.quiet, json=context.json):
+        run_pip_install(
+            context.target_prefix,
+            args=[pkg["url"] for pkg in pkgs.values()],
+            dry_run=context.dry_run,
+            quiet=context.quiet,
+            verbosity=context.verbosity,
+            force_reinstall=context.force_reinstall,
+            yes=context.always_yes,
+            check=True,
+        )
+
+    with Spinner("Verifying PyPI transaction", enabled=not context.quiet, json=context.json):
+        on_error_dict = {"disabled": "ignore", "warn": "warn", "enabled": "error"}
+        on_error = on_error_dict.get(context.safety_checks, "warn")
+        _verify_pypi_transaction(context.target_prefix, pkgs, on_error=on_error)
+
+    return 0
+
+
+def _pypi_lines_from_paths(paths: Iterable[str] | None = None) -> list[str]:
+    if paths is None:
+        file_arg = context.raw_data["cmd_line"].get("file")
+        if file_arg is None:
+            return []
+        paths = file_arg.value(None)
+    lines = []
+    line_prefix = PyPIDistribution._line_prefix
+    for path in paths:
+        path = path.value(None)
+        try:
+            with open(path) as f:
+                for line in f:
+                    if line.startswith(line_prefix):
+                        lines.append(line[len(line_prefix) :])
+        except OSError as exc:
+            raise CondaFileIOError(f"Could not process {path}") from exc
+    return lines

conda_pypi/cli/list.py

@@ -0,0 +1,22 @@
+import sys
+from conda.base.context import context
+
+from .. import __version__
+from ..main import pypi_lines_for_explicit_lockfile
+
+
+def post_command(command: str):
+    if command != "list":
+        return
+    cmd_line = context.raw_data.get("cmd_line", {})
+    if "--explicit" not in sys.argv and not cmd_line.get("explicit").value(None):
+        return
+    if "--no-pip" in sys.argv or not cmd_line.get("pip"):
+        return
+    checksums = ("md5",) if ("--md5" in sys.argv or cmd_line.get("md5").value(None)) else None
+    to_print = pypi_lines_for_explicit_lockfile(context.target_prefix, checksums=checksums)
+    if to_print:
+        sys.stdout.flush()
+        print(f"# The following lines were added by conda-pypi v{__version__}")
+        print("# This is an experimental feature subject to change. Do not use in production.")
+        print(*to_print, sep="\n")

conda_pypi/cli.py → conda_pypi/cli/pip.py

@@ -20,7 +20,7 @@
 
 
 def configure_parser(parser: argparse.ArgumentParser):
-    from .dependencies import BACKENDS
+    from ..dependencies import BACKENDS
 
     add_parser_help(parser)
     add_parser_prefix(parser)
@@ -69,14 +69,14 @@ def configure_parser(parser: argparse.ArgumentParser):
 def execute(args: argparse.Namespace) -> int:
     from conda.common.io import Spinner
     from conda.models.match_spec import MatchSpec
-    from .dependencies import analyze_dependencies
-    from .main import (
+    from ..dependencies import analyze_dependencies
+    from ..main import (
         validate_target_env,
         ensure_externally_managed,
         run_conda_install,
         run_pip_install,
     )
-    from .utils import get_prefix
+    from ..utils import get_prefix
 
     prefix = get_prefix(args.prefix, args.name)
     packages_not_installed = validate_target_env(prefix, args.packages)
@@ -150,7 +150,7 @@ def execute(args: argparse.Namespace) -> int:
     if pypi_specs:
         if not args.quiet or not args.json:
             print("Running pip install...")
-        retcode = run_pip_install(
+        process = run_pip_install(
             prefix,
             pypi_specs,
             dry_run=args.dry_run,
@@ -159,8 +159,8 @@ def execute(args: argparse.Namespace) -> int:
             force_reinstall=args.force_reinstall,
             yes=args.yes,
         )
-        if retcode:
-            return retcode
+        if process.returncode:
+            return process.returncode
         if os.environ.get("CONDA_BUILD_STATE") != "BUILD":
             ensure_externally_managed(prefix)
     return 0

conda_pypi/dependencies/pip.py

@@ -1,15 +1,9 @@
 from __future__ import annotations
 
 import json
-import os
 from logging import getLogger
 from collections import defaultdict
-from subprocess import run
-from tempfile import NamedTemporaryFile
-
-from conda.exceptions import CondaError
-
-from ..utils import get_env_python
+from ..main import dry_run_pip_json
 
 logger = getLogger(f"conda.{__name__}")
 
@@ -19,43 +13,7 @@ def _analyze_with_pip(
     prefix: str | None = None,
     force_reinstall: bool = False,
 ) -> tuple[dict[str, list[str]], dict[str, list[str]]]:
-    # pip can output to stdout via `--report -` (dash), but this
-    # creates issues on Windows due to undecodable characters on some
-    # project descriptions (e.g. charset-normalizer, amusingly), which
-    # makes pip crash internally. Probably a bug on their end.
-    # So we use a temporary file instead to work with bytes.
-    json_output = NamedTemporaryFile(suffix=".json", delete=False)
-    json_output.close()  # Prevent access errors on Windows
-
-    cmd = [
-        str(get_env_python(prefix)),
-        "-mpip",
-        "install",
-        "--dry-run",
-        "--ignore-installed",
-        *(("--force-reinstall",) if force_reinstall else ()),
-        "--report",
-        json_output.name,
-        *packages,
-    ]
-    process = run(cmd, capture_output=True, text=True)
-    if process.returncode != 0:
-        raise CondaError(
-            f"Failed to analyze dependencies with pip:\n"
-            f"  command: {' '.join(cmd)}\n"
-            f"  exit code: {process.returncode}\n"
-            f"  stderr:\n{process.stderr}\n"
-            f"  stdout:\n{process.stdout}\n"
-        )
-    logger.debug("pip (%s) provided the following report:\n%s", " ".join(cmd), process.stdout)
-
-    with open(json_output.name, "rb") as f:
-        # We need binary mode because the JSON output might
-        # contain weird unicode stuff (as part of the project
-        # description or README).
-        report = json.loads(f.read())
-    os.unlink(json_output.name)
-
+    report = dry_run_pip_json(("--prefix", prefix, *packages), force_reinstall)
     deps_from_pip = defaultdict(list)
     conda_deps = defaultdict(list)
     for item in report["install"]: