Process Module

Process execution and result handling utilities for subprocess management.

Overview

The process module provides low-level utilities for executing subprocesses and handling their results. It wraps Python's subprocess module with a more convenient API and structured result objects.

Key Components:

• ProcessResult - Structured result object containing exit code, stdout, stderr, and execution duration
• run_process() - Execute a subprocess and return a ProcessResult
• ensure_tool_available() - Verify an external tool is installed and accessible

Common Usage Patterns

Basic Process Execution

from shannot.process import run_process

# Simple execution
result = run_process(["ls", "-la", "/tmp"])
print(result.stdout)

# Check exit status
if result.succeeded():
    print("Command succeeded")
else:
    print(f"Failed with exit code {result.returncode}")

Error Handling

# Automatic error checking
try:
    result = run_process(["false"], check=True)
except subprocess.CalledProcessError as e:
    print(f"Command failed: {e}")

# Manual error checking
result = run_process(["cat", "/missing"], check=False)
if not result.succeeded():
    print(f"Error: {result.stderr}")
    print(f"Exit code: {result.returncode}")

Execution with Options

from pathlib import Path

# Custom working directory
result = run_process(
    ["pwd"],
    cwd="/tmp"
)

# Custom environment
result = run_process(
    ["env"],
    env={"CUSTOM_VAR": "value", "PATH": "/usr/bin"}
)

# Timeout
result = run_process(
    ["sleep", "10"],
    timeout=5.0  # Raises TimeoutExpired after 5 seconds
)

# Print command before execution
result = run_process(
    ["ls", "/"],
    print_command=True  # Prints: + ls /
)

Tool Availability Checking

from shannot.process import ensure_tool_available
from pathlib import Path

# Check if a tool exists
try:
    bwrap_path = ensure_tool_available("bwrap", search_path=True)
    print(f"Found bwrap at: {bwrap_path}")
except FileNotFoundError as e:
    print(f"Tool not found: {e}")

# Check specific path
ensure_tool_available(Path("/usr/bin/bwrap"), search_path=False)

Result Inspection

result = run_process(["du", "-sh", "/var/log"])

# Access result properties
print(f"Command: {' '.join(result.command)}")
print(f"Exit code: {result.returncode}")
print(f"Duration: {result.duration:.3f} seconds")
print(f"Output length: {len(result.stdout)} bytes")

# Check success
if result.succeeded():
    for line in result.stdout.splitlines():
        print(line)

Related Documentation

• Sandbox Module - Uses process utilities for command execution
• Execution Module - High-level execution interface
• Python API Guide - Integration examples

API Reference

process

Classes

ProcessResult dataclass

Represents the outcome of a subprocess invocation.

Source code in shannot/process.py

@dataclass
class ProcessResult:
    """Represents the outcome of a subprocess invocation."""

    command: tuple[str, ...]
    returncode: int
    stdout: str
    stderr: str
    duration: float

    def succeeded(self) -> bool:
        """Return True when the underlying process exited with status 0."""
        return self.returncode == 0

Functions

succeeded()

Return True when the underlying process exited with status 0.

Source code in shannot/process.py

def succeeded(self) -> bool:
    """Return True when the underlying process exited with status 0."""
    return self.returncode == 0

Functions

run_process(args, *, cwd=None, env=None, check=False, capture_output=True, timeout=None, print_command=False)

Execute args with subprocess.run and return a structured result.

Args: args: Command and arguments to execute. cwd: Optional working directory. env: Optional environment overrides. check: When True, re-raises CalledProcessError on non-zero exit. capture_output: When True, captures stdout/stderr into the result. timeout: Optional timeout in seconds. print_command: When True, echoes the command before execution.

Returns: ProcessResult with stdout/stderr always normalised to text.

Source code in shannot/process.py

def run_process(
    args: Sequence[str],
    *,
    cwd: Path | str | None = None,
    env: Mapping[str, str] | None = None,
    check: bool = False,
    capture_output: bool = True,
    timeout: float | None = None,
    print_command: bool = False,
) -> ProcessResult:
    """Execute ``args`` with ``subprocess.run`` and return a structured result.

    Args:
        args: Command and arguments to execute.
        cwd: Optional working directory.
        env: Optional environment overrides.
        check: When True, re-raises ``CalledProcessError`` on non-zero exit.
        capture_output: When True, captures stdout/stderr into the result.
        timeout: Optional timeout in seconds.
        print_command: When True, echoes the command before execution.

    Returns:
        ProcessResult with stdout/stderr always normalised to text.
    """
    if print_command:
        print("+", " ".join(str(part) for part in args))

    start = time.monotonic()
    cwd_param = str(cwd) if cwd is not None else None
    env_param = dict(env) if env is not None else None
    timeout_param = float(timeout) if timeout is not None else None
    try:
        if capture_output:
            completed = subprocess.run(
                list(args),
                cwd=cwd_param,
                env=env_param,
                check=check,
                capture_output=True,
                text=True,
                timeout=timeout_param,
            )
            stdout = completed.stdout or ""
            stderr = completed.stderr or ""
        else:
            completed = subprocess.run(
                list(args),
                cwd=cwd_param,
                env=env_param,
                check=check,
                capture_output=False,
                timeout=timeout_param,
            )
            stdout = ""
            stderr = ""
        return ProcessResult(
            command=tuple(args),
            returncode=completed.returncode,
            stdout=stdout,
            stderr=stderr,
            duration=time.monotonic() - start,
        )
    except subprocess.TimeoutExpired as exc:
        stdout = _decode_stream(getattr(exc, "stdout", None))
        stderr = _decode_stream(getattr(exc, "stderr", None) or getattr(exc, "output", None))
        timeout_seconds = timeout_param
        timeout_message = (
            f"Command timed out after {timeout_seconds} seconds"
            if timeout_seconds is not None
            else "Command timed out"
        )
        return ProcessResult(
            command=tuple(args),
            returncode=124,
            stdout=stdout,
            stderr=stderr or timeout_message,
            duration=time.monotonic() - start,
        )
    except subprocess.CalledProcessError as exc:
        stdout_raw = cast(bytes | str | None, getattr(exc, "stdout", None))
        stderr_raw = cast(bytes | str | None, getattr(exc, "stderr", None))
        stdout = _decode_stream(stdout_raw)
        stderr = _decode_stream(stderr_raw)
        return ProcessResult(
            command=tuple(args),
            returncode=exc.returncode,
            stdout=stdout,
            stderr=stderr,
            duration=time.monotonic() - start,
        )

ensure_tool_available(executable)

Ensure the given executable is present on PATH and return its resolved Path.

Source code in shannot/process.py

def ensure_tool_available(executable: str) -> Path:
    """Ensure the given executable is present on PATH and return its resolved Path."""
    resolved = shutil.which(executable)
    if resolved is None:
        raise FileNotFoundError(f"Required executable '{executable}' was not found in PATH")
    return Path(resolved)