feat(zfs): enhance command handling with run_zfs and run_zpool functions

python/zfs/command.py

@@ -0,0 +1,90 @@
+"""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)