1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
|
"""Subprocess runner and logging utilities."""
from __future__ import annotations
import os
import subprocess
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Iterable, Mapping
@dataclass
class RunResult:
cmd: list[str]
returncode: int
stdout: str
stderr: str
def log_line(level: str, scope: str, message: str) -> None:
ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
print(f"{ts} {level:<5} {scope}: {message}")
def run(
cmd: Iterable[str],
*,
log_file: str | None = None,
env: Mapping[str, str] | None = None,
allow_fail: bool = False,
dry_run: bool = False,
) -> RunResult:
"""Run a command and optionally tee output to a log file."""
cmd_list = list(cmd)
if dry_run:
if log_file:
with open(log_file, "a", encoding="utf-8") as fh:
fh.write(f"[dry-run] {' '.join(cmd_list)}\n")
return RunResult(cmd=cmd_list, returncode=0, stdout="", stderr="")
env_map = os.environ.copy()
if env:
env_map.update(env)
proc = subprocess.Popen(
cmd_list,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
env=env_map,
)
stdout, stderr = proc.communicate()
if log_file:
with open(log_file, "a", encoding="utf-8") as fh:
if stdout:
fh.write(stdout)
if stderr:
fh.write(stderr)
if proc.returncode != 0 and not allow_fail:
detail = ""
if stderr:
detail = f" stderr: {stderr.strip()}"
raise RuntimeError(
f"command failed: {' '.join(cmd_list)} (exit {proc.returncode}){detail}"
)
return RunResult(cmd=cmd_list, returncode=proc.returncode, stdout=stdout, stderr=stderr)
|
