cmd.py refactor part 4

pkgs/clan-cli/clan_cli/cmd.py

@@ -55,6 +55,7 @@ def run(
     cwd: Path = Path.cwd(),
     log: Log = Log.STDERR,
     check: bool = True,
+    error_msg: str | None = None,
 ) -> CmdOut:
     # Start the subprocess
     process = subprocess.Popen(
@@ -76,6 +77,7 @@ def run(
         cwd=cwd,
         command=shlex.join(cmd),
         returncode=process.returncode,
+        msg=error_msg,
     )
 
     if check and rc != 0:

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

@@ -4,12 +4,11 @@ import json
 import logging
 import os
 import re
-import shlex
-import subprocess
 import sys
 from pathlib import Path
 from typing import Any, get_origin
 
+from clan_cli.cmd import run
 from clan_cli.dirs import machine_settings_file
 from clan_cli.errors import ClanError
 from clan_cli.git import commit_file
@@ -117,15 +116,11 @@ def options_for_machine(
         f"{clan_dir}#nixosConfigurations.{machine_name}.config.clanCore.optionsNix"
     )
     cmd = nix_eval(flags=flags)
-    proc = subprocess.run(
+    proc = run(
         cmd,
-        stdout=subprocess.PIPE,
+        error_msg=f"Failed to read options for machine {machine_name}",
-        text=True,
     )
-    if proc.returncode != 0:
+
-        raise ClanError(
-            f"Failed to read options for machine {machine_name}:\n{shlex.join(cmd)}\nexit with {proc.returncode}"
-        )
     return json.loads(proc.stdout)
 
 
@@ -141,11 +136,8 @@ def read_machine_option_value(
             f"{clan_dir}#nixosConfigurations.{machine_name}.config.{option}",
         ],
     )
-    proc = subprocess.run(cmd, stdout=subprocess.PIPE, text=True)
+    proc = run(cmd, error_msg=f"Failed to read option {option}")
-    if proc.returncode != 0:
+
-        raise ClanError(
-            f"Failed to read option {option}:\n{shlex.join(cmd)}\nexit with {proc.returncode}"
-        )
     value = json.loads(proc.stdout)
     # print the value so that the output can be copied and fed as an input.
     # for example a list should be displayed as space separated values surrounded by quotes.

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

@@ -1,10 +1,9 @@
 import json
 import os
-import subprocess
-import sys
 from pathlib import Path
 from tempfile import NamedTemporaryFile
 
+from clan_cli.cmd import run
 from clan_cli.dirs import nixpkgs_source
 from clan_cli.errors import ClanError, ClanHttpError
 from clan_cli.nix import nix_eval
@@ -25,7 +24,7 @@ def machine_schema(
         clan_machine_settings_file.seek(0)
         env["CLAN_MACHINE_SETTINGS_FILE"] = clan_machine_settings_file.name
         # ensure that the requested clanImports exist
-        proc = subprocess.run(
+        proc = run(
             nix_eval(
                 flags=[
                     "--impure",
@@ -47,13 +46,11 @@ def machine_schema(
                     """,
                 ]
             ),
-            capture_output=True,
-            text=True,
             cwd=flake_dir,
             env=env,
+            check=False,
         )
         if proc.returncode != 0:
-            print(proc.stderr, file=sys.stderr)
             raise ClanHttpError(
                 status_code=400,
                 msg=f"Failed to check clanImports for existence:\n{proc.stderr}",
@@ -65,7 +62,7 @@ def machine_schema(
             )
 
         # get the schema
-        proc = subprocess.run(
+        proc = run(
             nix_eval(
                 flags=[
                     "--impure",
@@ -100,12 +97,10 @@ def machine_schema(
                     """,
                 ],
             ),
-            capture_output=True,
+            check=False,
-            text=True,
             cwd=flake_dir,
             env=env,
         )
     if proc.returncode != 0:
-        print(proc.stderr, file=sys.stderr)
         raise ClanError(f"Failed to read schema:\n{proc.stderr}")
     return json.loads(proc.stdout)

pkgs/clan-cli/clan_cli/errors.py

@@ -8,9 +8,11 @@ class CmdOut(NamedTuple):
     cwd: Path
     command: str
     returncode: int
+    msg: str | None = None
 
     def __str__(self) -> str:
         return f"""
+Message: {self.msg}
 Working Directory: '{self.cwd}'
 Return Code: {self.returncode}
 =================== Command ===================

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

@@ -1,11 +1,9 @@
 import argparse
 import json
 import logging
-import shlex
-import subprocess
 from pathlib import Path
 
-from ..errors import ClanError
+from ..cmd import run
 from ..nix import nix_config, nix_eval
 
 log = logging.getLogger(__name__)
@@ -22,17 +20,8 @@ def list_machines(flake_url: Path | str) -> list[str]:
             "--json",
         ]
     )
-    proc = subprocess.run(cmd, text=True, stdout=subprocess.PIPE)
+    proc = run(cmd)
-    assert proc.stdout is not None
+
-    if proc.returncode != 0:
-        raise ClanError(
-            f"""
-command: {shlex.join(cmd)}
-exit code: {proc.returncode}
-stdout:
-{proc.stdout}
-"""
-        )
     res = proc.stdout.strip()
     return json.loads(res)
 

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

@@ -1,6 +1,5 @@
 import json
 import os
-import subprocess
 import sys
 from pathlib import Path
 
@@ -70,10 +69,10 @@ class Machine:
         )  # TODO do this in the clanCore module
         env["SECRETS_DIR"] = str(secrets_dir)
         print(f"uploading secrets... {self.upload_secrets}")
-        proc = subprocess.run(
+        proc = run(
             [self.upload_secrets],
             env=env,
-            text=True,
+            check=False,
         )
 
         if proc.returncode == 23:

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

@@ -1,11 +1,9 @@
 import argparse
 import logging
 import os
-import subprocess
 import sys
 
-from clan_cli.errors import ClanError
+from ..cmd import run
-
 from ..machines.machines import Machine
 
 log = logging.getLogger(__name__)
@@ -17,15 +15,13 @@ def generate_secrets(machine: Machine) -> None:
     env["PYTHONPATH"] = ":".join(sys.path)  # TODO do this in the clanCore module
 
     print(f"generating secrets... {machine.generate_secrets}")
-    proc = subprocess.run(
+    run(
         [machine.generate_secrets],
         env=env,
+        error_msg="failed to generate secrets",
     )
 
-    if proc.returncode != 0:
+    print("successfully generated secrets")
-        raise ClanError("failed to generate secrets")
-    else:
-        print("successfully generated secrets")
 
 
 def generate_command(args: argparse.Namespace) -> None: