pc-backup/pc_backup/podman.py

import subprocess
from pathlib import Path


class Podman:
    @classmethod
    def run(
        cls,
        image: str,
        name: str,
        *,
        hostname: str,
        env: list,
        volumes: list[str],
        # TODO: Actually a list of Secret but creates a circular dependency
        secrets: list,
        ssh_auth_sock: Path | None = None,
        detach=True,
    ):
        args = ["run"]

        args += ["-h", hostname]
        args += ["--name", name]

        for e in env:
            args += ["-e", e]

        if ssh_auth_sock:
            args += ["-e", "SSH_AUTH_SOCK"]
            volumes += [f"{ssh_auth_sock}:{ssh_auth_sock}:Z"]

        args += ["--security-opt=label=disable"]

        if detach:
            args += ["--detach"]

        args += [a for vol in volumes for a in ["-v", vol]]
        args += [a for s in secrets for a in ["--secret", f"{s.name},mode=0{s.mode:o}"]]

        args += [image]

        cls._call(args)

    @classmethod
    def rm(cls, name, *, force=True):
        args = ["rm"]
        if force:
            args += ["-f"]
        args += [name]
        cls._call(args)

    @classmethod
    def exec(cls, name, *cmd):
        args = ["exec", "-ti", name] + list(cmd)
        cls._call(args)

    @classmethod
    def secret_create(
        cls,
        name: str,
        *,
        replace=True,
        value: bytes | None = None,
        host_path: Path | None = None,
    ):
        if value and host_path:
            raise ValueError("both value and host_path can not be set")

        args = ["secret", "create"]
        kwargs = {}

        if replace:
            args += ["--replace"]

        args += [name]

        if value is not None:
            args += ["-"]
            kwargs["input"] = value
        elif host_path is not None:
            args += [host_path]

        cls._call(args, **kwargs)

    @classmethod
    def _call(cls, args: list, **kwargs):
        args = ["podman"] + args
        print(f"Executing `{" ".join(args)}`")
        out = subprocess.check_output(args, **kwargs)
        print(out.decode())

    @classmethod
    def ps(cls):
        out = subprocess.run(["podman", "ps"], capture_output=True, check=True)

    @classmethod
    def machine_is_running(cls) -> bool:
        try:
            cls.ps()
        except subprocess.CalledProcessError as e:
            if "Cannot connect to Podman" in e.stderr.decode():
                return False

        return True

    @classmethod
    def machine_start(cls):
        cls._call(["machine", "start"])

    @classmethod
    def _call(cls, args: list, **kwargs):
        args = ["podman"] + args
        print(f"Executing `{" ".join(args)}`")

        def run_and_output():
            out = subprocess.check_output(args, **kwargs)
            print(out.decode())

        try:
            run_and_output()
        except subprocess.CalledProcessError as e:
            if not cls.machine_is_running:
                print("podman not running, starting it")
                cls.machine_start()
                run_and_output()
            else:
                raise e