home-services/composectl.py

#!/usr/bin/env python3
"""Runtime-agnostic compose deployment helper for home services.

This lightweight CLI keeps compose files as the source of truth and selects
Docker or Podman as the backend at runtime.
"""

from __future__ import annotations

import argparse
import shutil
import subprocess
import sys
from pathlib import Path


def detect_runtime(preferred: str | None = None) -> str:
    """Return a supported runtime name.

    The preferred runtime is honored when available; otherwise the first
    installed runtime from the standard Docker/Podman lookup order is used.
    """
    if preferred and shutil.which(preferred):
        return preferred

    for runtime in ("docker", "podman"):
        if shutil.which(runtime):
            return runtime

    raise RuntimeError("Neither Docker nor Podman is installed or available on PATH")


def build_compose_command(
    runtime: str,
    compose_file: Path,
    project_name: str,
    action: str,
    detached: bool = True,
) -> list[str]:
    """Build a compose command for the chosen runtime."""
    base = [runtime, "compose", "-f", str(compose_file), "-p", project_name]

    if action == "up":
        return [*base, "up", "-d"]
    if action == "pull":
        return [*base, "pull"]
    if action == "down":
        return [*base, "down"]
    if action == "status":
        return [*base, "ps"]

    raise ValueError(f"Unsupported action '{action}'")


def discover_compose_files(root: Path) -> list[Path]:
    """Return compose files under the services directory."""
    services_dir = root / "services"
    if not services_dir.exists():
        return []

    return sorted(path for path in services_dir.rglob("compose.yaml") if path.is_file())


def discover_stack_names(root: Path) -> list[str]:
    """Infer stack names from the services directory layout."""
    services_dir = root / "services"
    if not services_dir.exists():
        return []

    return sorted(path.name for path in services_dir.iterdir() if path.is_dir())


def build_update_commands(runtime: str, compose_file: Path, project_name: str) -> list[list[str]]:
    """Return the maintenance sequence used for a full stack refresh."""
    return [
        build_compose_command(runtime=runtime, compose_file=compose_file, project_name=project_name, action="pull"),
        build_compose_command(runtime=runtime, compose_file=compose_file, project_name=project_name, action="down"),
        build_compose_command(runtime=runtime, compose_file=compose_file, project_name=project_name, action="up"),
        [runtime, "image", "prune", "-f"],
    ]


def run_command(command: list[str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(command, check=False, text=True, capture_output=True)


def main() -> int:
    parser = argparse.ArgumentParser(description="Manage compose stacks with Docker or Podman")
    parser.add_argument("action", choices=["pull", "up", "down", "prune", "status", "refresh"])
    parser.add_argument("--stack", default=None)
    parser.add_argument("--compose-file", type=Path, default=None)
    parser.add_argument("--project-name", default=None)
    parser.add_argument("--runtime", choices=["docker", "podman"], default=None)
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--all", action="store_true", help="Run the requested action for every configured stack")

    args = parser.parse_args()

    runtime = detect_runtime(args.runtime)

    if args.action == "prune":
        command = [runtime, "image", "prune", "-f"]
        if args.dry_run:
            print(" ".join(command))
            return 0

        result = run_command(command)
        if result.stdout:
            print(result.stdout, end="")
        if result.stderr:
            print(result.stderr, end="", file=sys.stderr)
        return result.returncode

    stack_names = discover_stack_names(Path(".")) if args.all else [args.stack or "media"]
    failure_count = 0

    for stack_name in stack_names:
        compose_file = args.compose_file or Path(f"services/{stack_name}/compose.yaml")
        project_name = args.project_name or f"{stack_name}_services"

        if args.action == "refresh":
            commands = build_update_commands(runtime=runtime, compose_file=compose_file, project_name=project_name)
            if args.dry_run:
                for command in commands:
                    print(" ".join(command))
                continue

            for command in commands:
                result = run_command(command)
                if result.stdout:
                    print(result.stdout, end="")
                if result.stderr:
                    print(result.stderr, end="", file=sys.stderr)
                if result.returncode != 0:
                    failure_count += 1
            continue

        command = build_compose_command(
            runtime=runtime,
            compose_file=compose_file,
            project_name=project_name,
            action=args.action,
        )

        if args.dry_run:
            print(" ".join(command))
            continue

        result = run_command(command)
        if result.stdout:
            print(result.stdout, end="")
        if result.stderr:
            print(result.stderr, end="", file=sys.stderr)
        if result.returncode != 0:
            failure_count += 1

    return 1 if failure_count else 0


if __name__ == "__main__":
    raise SystemExit(main())