forked from clan/clan-core
Added ClanCmdError. cmd.run now returns this error
This commit is contained in:
parent
626e5558f3
commit
699515ca89
@ -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(
|
||||
f"""
|
||||
command: {shlex.join(cmd)}
|
||||
working directory: {cwd}
|
||||
exit code: {rc}
|
||||
stderr:
|
||||
{stderr_buf}
|
||||
stdout:
|
||||
{stdout_buf}
|
||||
"""
|
||||
)
|
||||
raise ClanCmdError(cmd_out)
|
||||
|
||||
return CmdOut(stdout_buf, stderr_buf, cwd=cwd)
|
||||
|
||||
|
||||
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)
|
||||
return cmd_out
|
||||
|
@ -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__()
|
||||
|
@ -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
|
||||
|
Loading…
Reference in New Issue
Block a user