clan-cli: replace run_cmd() with env_for_programs()

Instead of wrapping a cmd with nix-shell, env_for_programs() returns an `env` containing the right PATH to be passed to a subprocess.run invocation.

programs which are not statically passed via CLAN_STATIC_DEPS, will be retrieved via `nix build` and added to the PATH

There is now also a cache which will prevent programs being evaluated multiple ties during the life cycle of one process

pkgs/clan-cli/clan_cli/git.py

@@ -2,7 +2,7 @@ from pathlib import Path
 
 # from clan_cli.dirs import find_git_repo_root
 from clan_cli.errors import ClanError
-from clan_cli.nix import run_cmd
+from clan_cli.nix import env_for_programs
 
 from .cmd import Log, run
 from .locked_open import locked_open
@@ -60,43 +60,29 @@ def _commit_file_to_git(
     """
     with locked_open(repo_dir / ".git" / "clan.lock", "w+"):
         for file_path in file_paths:
-            cmd = run_cmd(
-                ["git"],
-                ["git", "-C", str(repo_dir), "add", str(file_path)],
-            )
             # add the file to the git index
-
             run(
-                cmd,
+                ["git", "-C", str(repo_dir), "add", str(file_path)],
+                env=env_for_programs(["nixpkgs#git"]),
                 log=Log.BOTH,
                 error_msg=f"Failed to add {file_path} file to git index",
             )
 
-        # check if there is a diff
-        cmd = run_cmd(
-            ["git"],
+        result = run(
             ["git", "-C", str(repo_dir), "diff", "--cached", "--exit-code"]
             + [str(file_path) for file_path in file_paths],
+            env=env_for_programs(["nixpkgs#git"]),
+            check=False,
+            cwd=repo_dir,
+            error_msg=f"Failed to check if there is a diff in the git repository {repo_dir}",
         )
-        result = run(cmd, check=False, cwd=repo_dir)
         # if there is no diff, return
         if result.returncode == 0:
             return
 
-        # commit only that file
-        cmd = run_cmd(
-            ["git"],
-            [
-                "git",
-                "-C",
-                str(repo_dir),
-                "commit",
-                "-m",
-                commit_message,
-            ]
-            + [str(file_path) for file_path in file_paths],
-        )
-
         run(
-            cmd, error_msg=f"Failed to commit {file_paths} to git repository {repo_dir}"
+            ["git", "-C", str(repo_dir), "commit", "-m", commit_message]
+            + [str(file_path) for file_path in file_paths],
+            env=env_for_programs(["nixpkgs#git"]),
+            error_msg=f"Failed to commit {file_paths} to git repository {repo_dir}",
         )

pkgs/clan-cli/clan_cli/nix/__init__.py

@@ -1,8 +1,9 @@
 import json
 import os
+import subprocess
 import tempfile
 from pathlib import Path
-from typing import Any
+from typing import Any, ClassVar
 
 from ..cmd import run, run_no_stdout
 from ..dirs import nixpkgs_flake, nixpkgs_source
@@ -113,47 +114,80 @@ def nix_shell(packages: list[str], cmd: list[str]) -> list[str]:
     ]
 
 
-# lazy loads list of allowed and static programs
+# Cache for requested dependencies
 class Programs:
-    allowed_programs = None
-    static_programs = None
+    _allowed_programs = None
+    _clan_static_deps = None
+    _cached_paths: ClassVar[dict[str, list[str]]] = {}
+    _to_resolve: ClassVar[list[str]] = []
 
     @classmethod
-    def is_allowed(cls: type["Programs"], program: str) -> bool:
-        if cls.allowed_programs is None:
-            with open(Path(__file__).parent / "allowed-programs.json") as f:
-                cls.allowed_programs = json.load(f)
-        return program in cls.allowed_programs
+    def add_program(cls: type["Programs"], flake_ref: str) -> None:
+        if "#" not in flake_ref:
+            raise ValueError(
+                f"Program {flake_ref} is not a flake reference. Please use the format 'nixpkgs#{flake_ref}'"
+            )
+        if cls._allowed_programs is None:
+            cls._allowed_programs = set(
+                json.loads(
+                    (Path(__file__).parent / "allowed-programs.json").read_text()
+                )
+            )
+        if cls._clan_static_deps is None:
+            cls._clan_static_deps = json.loads(os.environ.get("CLAN_STATIC_DEPS", "{}"))
+        if flake_ref in cls._cached_paths:
+            return
+        if flake_ref.startswith("nixpkgs#"):
+            name = flake_ref.split("#")[1]
+            if name not in cls._allowed_programs:
+                raise ValueError(
+                    f"Program {name} is not allowed as it is not in the allowed-programs.json file."
+                )
+            if name in cls._clan_static_deps:
+                cls._cached_paths[flake_ref] = [cls._clan_static_deps[name]]
+                return
+        cls._to_resolve.append(flake_ref)
 
     @classmethod
-    def is_static(cls: type["Programs"], program: str) -> bool:
-        """
-        Determines if a program is statically shipped with this clan distribution
-        """
-        if cls.static_programs is None:
-            cls.static_programs = os.environ.get("CLAN_STATIC_PROGRAMS", "").split(":")
-        return program in cls.static_programs
+    # TODO: optimize via multiprocessing
+    def resolve_all(cls: type["Programs"]) -> None:
+        for flake_ref in cls._to_resolve:
+            if flake_ref in cls._cached_paths:
+                continue
+            build = subprocess.run(
+                nix_command(
+                    [
+                        "build",
+                        "--inputs-from",
+                        f"{nixpkgs_flake()!s}",
+                        "--no-link",
+                        "--print-out-paths",
+                        flake_ref,
+                    ]
+                ),
+                capture_output=True,
+                check=True,
+            )
+            paths = build.stdout.decode().strip().splitlines()
+            cls._cached_paths[flake_ref] = list(map(lambda path: path + "/bin", paths))
+        cls._to_resolve = []
+
+    @classmethod
+    def bin_paths(cls: type["Programs"], names_or_refs: list[str]) -> list[str]:
+        for name_or_ref in names_or_refs:
+            cls.add_program(name_or_ref)
+        breakpoint()
+        cls.resolve_all()
+        paths = []
+        for name_or_ref in names_or_refs:
+            paths.extend(cls._cached_paths[name_or_ref])
+        return paths
 
 
-# Alternative implementation of nix_shell() to replace nix_shell() at some point
-#   Features:
-#     - allow list for programs (need to be specified in allowed-programs.json)
-#     - be abe to compute a closure of all deps for testing
-#     - build clan distributions that ship some or all packages (eg. clan-cli-full)
-def run_cmd(programs: list[str], cmd: list[str]) -> list[str]:
-    for program in programs:
-        if not Programs.is_allowed(program):
-            raise ValueError(f"Program not allowed: {program}")
-    if os.environ.get("IN_NIX_SANDBOX"):
-        return cmd
-    missing_packages = [
-        f"nixpkgs#{program}" for program in programs if not Programs.is_static(program)
-    ]
-    if not missing_packages:
-        return cmd
-    return [
-        *nix_command(["shell", "--inputs-from", f"{nixpkgs_flake()!s}"]),
-        *missing_packages,
-        "-c",
-        *cmd,
-    ]
+def path_for_programs(packages: list[str]) -> str:
+    bin_paths = Programs.bin_paths(packages)
+    return ":".join(bin_paths)
+
+
+def env_for_programs(packages: list[str]) -> dict[str, str]:
+    return os.environ | {"PATH": path_for_programs(packages)}

pkgs/clan-cli/clan_cli/vars/generate.py

@@ -237,7 +237,7 @@ def _generate_vars_for_machine(
         for dep in dependencies:
             if dep not in graph:
                 raise ClanError(
-                    f"Generator {gen_name} has a dependency on {dep}, which does not exist"
+                    f"Generator {gen_name} has a dependency on {dep}, which does not exist for machine {machine.name}"
                 )
 
     # process generators in topological order

pkgs/clan-cli/default.nix

@@ -27,6 +27,8 @@ let
     argcomplete # Enables shell completions
   ];
 
+  clanStaticDeps = lib.genAttrs includedRuntimeDeps (name: lib.getBin (pkgs.${name}));
+
   # load nixpkgs runtime dependencies from a json file
   # This file represents an allow list at the same time that is checked by the run_cmd
   #   implementation in nix.py
@@ -99,16 +101,12 @@ python3.pkgs.buildPythonApplication {
       "${gitMinimal}/bin/git"
     ]
     # include selected runtime dependencies in the PATH
-    ++ lib.concatMap (p: [
-      "--prefix"
-      "PATH"
-      ":"
-      p
-    ]) includedRuntimeDeps
     ++ [
       "--set"
-      "CLAN_STATIC_PROGRAMS"
-      (lib.concatStringsSep ":" includedRuntimeDeps)
+      "CLAN_STATIC_DEPS"
+      (pkgs.writeText "static-deps.json" (
+        builtins.toJSON (lib.genAttrs includedRuntimeDeps (name: lib.getBin (pkgs.${name})))
+      ))
     ];
 
   nativeBuildInputs = [
@@ -162,7 +160,7 @@ python3.pkgs.buildPythonApplication {
   passthru.testDependencies = testDependencies;
   passthru.pythonWithTestDeps = pythonWithTestDeps;
   passthru.runtimeDependencies = runtimeDependencies;
-  passthru.runtimeDependenciesAsSet = runtimeDependenciesAsSet;
+  passthru.clanStaticDeps = clanStaticDeps;
 
   postInstall = ''
     cp -r ${nixpkgs'} $out/${python3.sitePackages}/clan_cli/nixpkgs

pkgs/clan-cli/shell.nix

@@ -1,5 +1,4 @@
 {
-  lib,
   nix-unit,
   clan-cli,
   clan-cli-full,
@@ -29,9 +28,7 @@ mkShell {
 
   PYTHONBREAKPOINT = "ipdb.set_trace";
 
-  CLAN_STATIC_PROGRAMS = lib.concatStringsSep ":" (
-    lib.attrNames clan-cli-full.passthru.runtimeDependenciesAsSet
-  );
+  CLAN_STATIC_DEPS = builtins.toJSON (clan-cli-full.passthru.clanStaticDeps);
 
   shellHook = ''
     export GIT_ROOT="$(git rev-parse --show-toplevel)"

pkgs/clan-cli/tests/test_nix.py

@@ -0,0 +1,9 @@
+import pytest
+
+@pytest.mark.impure
+def test_path_for_programs():
+    from clan_cli.nix import path_for_programs
+    from pathlib import Path
+    path = path_for_programs(["nixpkgs#bash"])
+    assert Path(path).exists()
+    assert path.startswith("/nix/store/")