Richie
48a7e3a54c
treefmt / nix fmt (pull_request) Successful in 6s
pytest / pytest (pull_request) Successful in 30s
test ebook search / test-ebook-search (pull_request) Successful in 36s
build_systems / build-brain (pull_request) Successful in 46s
build_systems / build-bob (pull_request) Successful in 47s
build_systems / build-rhapsody-in-green (pull_request) Successful in 58s
build_systems / build-jeeves (pull_request) Successful in 2m23s
pytest / pytest (push) Successful in 33s
test ebook search / test-ebook-search (push) Successful in 42s
build_systems / build-jeeves (push) Successful in 2m26s
treefmt / nix fmt (push) Successful in 5s
build_systems / build-brain (push) Successful in 9s
build_systems / build-bob (push) Successful in 40s
build_systems / build-rhapsody-in-green (push) Successful in 53s
91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
"""Running zfs and zpool commands.
|
|
|
|
One implementation shared by both, so the zpool side gets the same handling the
|
|
zfs side does: arguments passed as a list, streams kept apart, and failures
|
|
returned as data rather than guessed at by the caller.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CommandResult:
|
|
"""The outcome of a single zfs or zpool invocation."""
|
|
|
|
args: tuple[str, ...]
|
|
stdout: str
|
|
stderr: str
|
|
return_code: int
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
"""Whether the command reported success."""
|
|
return self.return_code == 0
|
|
|
|
@property
|
|
def message(self) -> str:
|
|
"""The most useful description of what went wrong."""
|
|
return (self.stderr or self.stdout).strip()
|
|
|
|
|
|
def run_command(*args: str) -> CommandResult:
|
|
"""Run a command, passing arguments as a list rather than a shell string.
|
|
|
|
Two things this buys over bash_wrapper. Arguments are never split on
|
|
whitespace, so a value containing a space arrives intact. And stdout stays
|
|
separate from stderr, so a warning on a successful command is never
|
|
mistaken for output, which bash_wrapper does whenever stderr is non-empty
|
|
regardless of the return code.
|
|
|
|
The encoding is pinned rather than using text=True, which would decode with
|
|
the locale encoding. These run from systemd units, where LANG is often
|
|
unset.
|
|
|
|
Args:
|
|
*args: The command and its arguments.
|
|
|
|
Returns:
|
|
CommandResult: The streams and return code, never raising on failure.
|
|
"""
|
|
completed = subprocess.run(list(args), capture_output=True, encoding="utf-8", check=False)
|
|
|
|
if completed.returncode != 0:
|
|
logger.debug(f"{' '.join(args)} exited {completed.returncode}: {completed.stderr.strip()}")
|
|
|
|
return CommandResult(
|
|
args=tuple(args),
|
|
stdout=completed.stdout,
|
|
stderr=completed.stderr,
|
|
return_code=completed.returncode,
|
|
)
|
|
|
|
|
|
def run_zfs(*args: str) -> CommandResult:
|
|
"""Run a zfs command.
|
|
|
|
Args:
|
|
*args: The arguments to pass to zfs.
|
|
|
|
Returns:
|
|
CommandResult: The streams and return code.
|
|
"""
|
|
return run_command("zfs", *args)
|
|
|
|
|
|
def run_zpool(*args: str) -> CommandResult:
|
|
"""Run a zpool command.
|
|
|
|
Args:
|
|
*args: The arguments to pass to zpool.
|
|
|
|
Returns:
|
|
CommandResult: The streams and return code.
|
|
"""
|
|
return run_command("zpool", *args)
|