Refactor subprocess to cmd.py part 2

pkgs/clan-cli/clan_cli/cmd.py

@@ -5,6 +5,7 @@ import shlex
 import subprocess
 import sys
 from collections.abc import Callable
+from enum import Enum
 from pathlib import Path
 from typing import IO, Any, NamedTuple
 
@@ -16,7 +17,7 @@ log = logging.getLogger(__name__)
 class CmdOut(NamedTuple):
     stdout: str
     stderr: str
-    cwd: Path | None = None
+    cwd: Path
 
 
 def handle_output(process: subprocess.Popen) -> tuple[str, str]:

pkgs/clan-cli/clan_cli/config/parsing.py

@@ -1,8 +1,8 @@
 import json
-import subprocess
 from pathlib import Path
 from typing import Any
 
+from ..cmd import run
 from ..errors import ClanError
 from ..nix import nix_eval
 
@@ -32,7 +32,7 @@ def schema_from_module_file(
     """
     # run the nix expression and parse the output as json
     cmd = nix_eval(["--expr", nix_expr])
-    proc = subprocess.run(cmd, stdout=subprocess.PIPE, check=True)
+    proc = run(cmd)
     return json.loads(proc.stdout)
 
 

pkgs/clan-cli/clan_cli/machines/install.py

@@ -1,9 +1,9 @@
 import argparse
-import subprocess
 from dataclasses import dataclass
 from pathlib import Path
 from tempfile import TemporaryDirectory
 
+from ..cmd import Log, run
 from ..machines.machines import Machine
 from ..nix import nix_shell
 from ..secrets.generate import generate_secrets
@@ -40,12 +40,12 @@ def install_nixos(machine: Machine, kexec: str | None = None) -> None:
             cmd += ["--kexec", kexec]
         cmd.append(target_host)
 
-        subprocess.run(
+        run(
             nix_shell(
                 ["nixpkgs#nixos-anywhere"],
                 cmd,
             ),
-            check=True,
+            log=Log.BOTH,
         )
 
 

pkgs/clan-cli/clan_cli/machines/machines.py

@@ -4,7 +4,7 @@ import subprocess
 import sys
 from pathlib import Path
 
-from ..errors import ClanError
+from ..cmd import run
 from ..nix import nix_build, nix_config, nix_eval
 from ..ssh import Host, parse_deployment_address
 
@@ -13,21 +13,14 @@ def build_machine_data(machine_name: str, clan_dir: Path) -> dict:
     config = nix_config()
     system = config["system"]
 
-    proc = subprocess.run(
+    proc = run(
         nix_build(
             [
                 f'{clan_dir}#clanInternals.machines."{system}"."{machine_name}".config.system.clan.deployment.file'
             ]
         ),
-        stdout=subprocess.PIPE,
-        check=True,
-        text=True,
     )
 
-    if proc.returncode != 0:
-        ClanError("failed to build machine data")
-        exit(1)
-
     return json.loads(Path(proc.stdout.strip()).read_text())
 
 
@@ -99,11 +92,8 @@ class Machine:
         if attr in self.eval_cache and not refresh:
             return self.eval_cache[attr]
 
-        output = subprocess.run(
+        output = run(
             nix_eval([f"path:{self.flake_dir}#{attr}"]),
-            stdout=subprocess.PIPE,
-            check=True,
-            text=True,
         ).stdout.strip()
         self.eval_cache[attr] = output
         return output
@@ -115,11 +105,8 @@ class Machine:
         """
         if attr in self.build_cache and not refresh:
             return self.build_cache[attr]
-        outpath = subprocess.run(
+        outpath = run(
             nix_build([f"path:{self.flake_dir}#{attr}"]),
-            stdout=subprocess.PIPE,
-            check=True,
-            text=True,
         ).stdout.strip()
         self.build_cache[attr] = Path(outpath)
         return Path(outpath)

pkgs/clan-cli/clan_cli/machines/update.py

@@ -4,6 +4,7 @@ import os
 import subprocess
 from pathlib import Path
 
+from ..cmd import run
 from ..errors import ClanError
 from ..machines.machines import Machine
 from ..nix import nix_build, nix_command, nix_config
@@ -79,11 +80,8 @@ def deploy_nixos(hosts: HostGroup, clan_dir: Path) -> None:
 def get_all_machines(clan_dir: Path) -> HostGroup:
     config = nix_config()
     system = config["system"]
-    machines_json = subprocess.run(
+    machines_json = run(
-        nix_build([f'{clan_dir}#clanInternals.all-machines-json."{system}"']),
+        nix_build([f'{clan_dir}#clanInternals.all-machines-json."{system}"'])
-        stdout=subprocess.PIPE,
-        check=True,
-        text=True,
     ).stdout
 
     machines = json.loads(Path(machines_json.rstrip()).read_text())

pkgs/clan-cli/clan_cli/nix.py

@@ -1,10 +1,10 @@
 import json
 import os
-import subprocess
 import tempfile
 from pathlib import Path
 from typing import Any
 
+from .cmd import run
 from .dirs import nixpkgs_flake, nixpkgs_source
 
 
@@ -55,7 +55,7 @@ def nix_build(flags: list[str], gcroot: Path | None = None) -> list[str]:
 
 def nix_config() -> dict[str, Any]:
     cmd = nix_command(["show-config", "--json"])
-    proc = subprocess.run(cmd, check=True, text=True, stdout=subprocess.PIPE)
+    proc = run(cmd)
     data = json.loads(proc.stdout)
     config = {}
     for key, value in data.items():
@@ -90,7 +90,7 @@ def nix_eval(flags: list[str]) -> list[str]:
 
 def nix_metadata(flake_url: str | Path) -> dict[str, Any]:
     cmd = nix_command(["flake", "metadata", "--json", f"{flake_url}"])
-    proc = subprocess.run(cmd, check=True, text=True, stdout=subprocess.PIPE)
+    proc = run(cmd)
     data = json.loads(proc.stdout)
     return data
 

pkgs/clan-vm-manager/clan_vm_manager/executor.py

@@ -111,8 +111,7 @@ def spawn(
 
     if not log_path.is_dir():
         raise ClanError(f"Log path {log_path} is not a directory")
-    if not log_path.exists():
+    log_path.mkdir(parents=True, exist_ok=True)
-        log_path.mkdir(parents=True)
 
     # Set names
     proc_name = f"MPExec:{func.__name__}"
@@ -128,8 +127,6 @@ def spawn(
     proc.start()
 
     # Print some information
-    assert proc.pid is not None
-
     cmd = f"tail -f {out_file}"
     print(f"Connect to stdout with: {cmd}")