Merge pull request 'Added ClanCmdError. cmd.run now returns this error' (#704) from Qubasa-main into main

pkgs/clan-cli/clan_cli/cmd.py

@@ -4,22 +4,15 @@ import select
 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
+from typing import IO, Any
 
-from .errors import ClanError
+from .errors import ClanCmdError, CmdOut
 
 log = logging.getLogger(__name__)
 
 
-class CmdOut(NamedTuple):
-    stdout: str
-    stderr: str
-    cwd: Path
-
-
 class Log(Enum):
     STDERR = 1
     STDOUT = 2
@@ -55,42 +48,36 @@ def handle_output(process: subprocess.Popen, log: Log) -> tuple[str, str]:
     return stdout_buf.decode("utf-8"), stderr_buf.decode("utf-8")
 
 
-def run(cmd: list[str], cwd: Path = Path.cwd(), log: Log = Log.STDERR) -> CmdOut:
+def run(
+    cmd: list[str],
+    *,
+    env: dict[str, str] | None = None,
+    cwd: Path = Path.cwd(),
+    log: Log = Log.STDERR,
+) -> CmdOut:
     # Start the subprocess
     process = subprocess.Popen(
-        cmd, cwd=str(cwd), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
+        cmd,
+        cwd=str(cwd),
+        env=env,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
     )
 
     stdout_buf, stderr_buf = handle_output(process, log)
 
     # Wait for the subprocess to finish
     rc = process.wait()
+    cmd_out = CmdOut(
+        stdout_buf,
+        stderr_buf,
+        cwd=cwd,
+        command=shlex.join(cmd),
+        returncode=process.returncode,
+    )
 
     if rc != 0:
-        raise ClanError(
+        raise ClanCmdError(cmd_out)
-            f"""
-command: {shlex.join(cmd)}
-working directory: {cwd}
-exit code: {rc}
-stderr:
-{stderr_buf}
-stdout:
-{stdout_buf}
-"""
-        )
 
-    return CmdOut(stdout_buf, stderr_buf, cwd=cwd)
+    return cmd_out
-
-
-def runforcli(func: Callable[..., dict[str, CmdOut]], *args: Any) -> None:
-    try:
-        res = func(*args)
-
-        for name, out in res.items():
-            if out.stderr:
-                print(f"{name}: {out.stderr}", end="")
-            if out.stdout:
-                print(f"{name}: {out.stdout}", end="")
-    except ClanError as e:
-        print(e)
-        exit(1)

pkgs/clan-cli/clan_cli/errors.py

@@ -1,3 +1,15 @@
+from pathlib import Path
+from typing import NamedTuple
+
+
+class CmdOut(NamedTuple):
+    stdout: str
+    stderr: str
+    cwd: Path
+    command: str
+    returncode: int
+
+
 class ClanError(Exception):
     """Base class for exceptions in this module."""
 
@@ -12,3 +24,11 @@ class ClanHttpError(ClanError):
         self.status_code = status_code
         self.msg = msg
         super().__init__(msg)
+
+
+class ClanCmdError(ClanError):
+    cmd: CmdOut
+
+    def __init__(self, cmd: CmdOut) -> None:
+        self.cmd = cmd
+        super().__init__()

pkgs/clan-cli/clan_cli/flakes/create.py

@@ -2,7 +2,7 @@
 import argparse
 from pathlib import Path
 
-from ..cmd import CmdOut, run, runforcli
+from ..cmd import CmdOut, run
 from ..errors import ClanError
 from ..nix import nix_command, nix_shell
 
@@ -24,7 +24,6 @@ def create_flake(directory: Path, url: str) -> dict[str, CmdOut]:
         ]
     )
     out = run(command, cwd=directory)
-    response["flake init"] = out
 
     command = nix_shell(["nixpkgs#git"], ["git", "init"])
     out = run(command, cwd=directory)
@@ -48,7 +47,7 @@ def create_flake(directory: Path, url: str) -> dict[str, CmdOut]:
 
 
 def create_flake_command(args: argparse.Namespace) -> None:
-    runforcli(create_flake, args.path, args.url)
+    create_flake(args.path, args.url)
 
 
 # takes a (sub)parser and configures it