feat: Add script to pull and redeploy Docker containers

docker_refresh.py

@@ -0,0 +1,185 @@
+#!/usr/bin/env python3
+# /// script
+# dependencies = []
+# ///
+
+import argparse
+from dataclasses import dataclass, field
+import json
+import logging
+from pathlib import Path
+import re
+import shlex
+import subprocess
+import sys
+from typing import Any
+
+
+@dataclass
+class Project:
+    name: str
+    directory: Path
+    files: tuple[Path, ...]
+    environment_files: tuple[Path, ...]
+    services: set[str] = field(default_factory=set)
+
+    def command(self) -> list[str]:
+        command = [
+            "docker", "compose", "--project-name", self.name,
+            "--project-directory", str(self.directory), "--profile", "*",
+        ]
+        for path in self.files:
+            command.extend(["--file", str(path)])
+        for path in self.environment_files:
+            command.extend(["--env-file", str(path)])
+        return command
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Pull and redeploy running Compose services with unpinned images.",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+        epilog="Digests and tags starting with a version number (e.g. v1.2.3 or 16-alpine) "
+        "are skipped. Run on the Compose host with the original files and environment available.",
+    )
+    parser.add_argument("projects", nargs="*", help="Only refresh these Compose projects; omit to refresh all eligible projects")
+    parser.add_argument("--dry-run", action="store_true", help="Discover services and log commands without changing deployments")
+    return parser.parse_args()
+
+
+def is_pinned(image: str) -> bool:
+    if "@" in image:
+        return True
+    final_component = image.rsplit("/", 1)[-1]
+    if ":" not in final_component:
+        return False
+    tag = final_component.rsplit(":", 1)[-1]
+    return re.match(r"^v?\d+(?:[._-]|$)", tag) is not None
+
+
+def inspect_containers() -> list[dict[str, Any]]:
+    result = subprocess.run(
+        args=["docker", "container", "ls", "--quiet"],
+        check=True, capture_output=True, text=True, timeout=60,
+    )
+    identifiers = result.stdout.split()
+    containers = []
+    for offset in range(0, len(identifiers), 100):
+        result = subprocess.run(
+            args=["docker", "container", "inspect", *identifiers[offset:offset + 100]],
+            check=True, capture_output=True, text=True, timeout=60,
+        )
+        value = json.loads(result.stdout)
+        if not isinstance(value, list):
+            raise ValueError("Docker returned an unexpected response")
+        containers.extend(value)
+    return containers
+
+
+def discover_projects(containers: list[dict[str, Any]], selected_projects: set[str]) -> list[Project]:
+    projects: dict[str, Project] = {}
+    for container in containers:
+        if not container.get("State", {}).get("Running"):
+            continue
+        config = container.get("Config") or {}
+        labels = config.get("Labels") or {}
+        name = labels.get("com.docker.compose.project")
+        service = labels.get("com.docker.compose.service")
+        if selected_projects and name not in selected_projects:
+            continue
+        if labels.get("com.docker.compose.oneoff", "").lower() == "true":
+            continue
+        image = config.get("Image") or ""
+        if not image:
+            logging.info("Skipping %s/%s: container image reference is missing", name, service)
+            continue
+        if is_pinned(image=image):
+            logging.warning("Skipping %s/%s: deployed image %s is pinned", name, service, image)
+            continue
+        if not name or not service:
+            logging.warning("Skipping %s: unpinned image %s is not a Compose service", container.get("Name"), image)
+            continue
+        directory = labels.get("com.docker.compose.project.working_dir")
+        files = labels.get("com.docker.compose.project.config_files")
+        if not directory or not files:
+            raise ValueError(f"Project {name}: missing Compose working directory or config file labels")
+        working_directory = Path(directory)
+        config_files = tuple(working_directory / path for path in files.split(",") if path)
+        environment_files = tuple(
+            working_directory / path
+            for path in labels.get("com.docker.compose.project.environment_file", "").split(",") if path
+        )
+        if not config_files:
+            raise ValueError(f"Project {name}: no Compose config files")
+        project = Project(name=name, directory=working_directory, files=config_files, environment_files=environment_files)
+        existing = projects.setdefault(name, project)
+        if (existing.directory, existing.files, existing.environment_files) != (project.directory, project.files, project.environment_files):
+            raise ValueError(f"Project {name}: containers have inconsistent Compose file labels")
+        existing.services.add(service)
+    return sorted(projects.values(), key=lambda project: project.name)
+
+
+def refresh_project(project: Project, dry_run: bool) -> None:
+    if not project.directory.is_dir():
+        raise ValueError(f"Project {project.name}: directory is unavailable: {project.directory}; run on the Compose host")
+    for path in (*project.files, *project.environment_files):
+        if not path.is_file():
+            raise ValueError(f"Project {project.name}: file is unavailable: {path}")
+    command = project.command()
+    result = subprocess.run(
+        args=[*command, "config", "--format", "json"], cwd=project.directory,
+        check=True, capture_output=True, text=True, timeout=60,
+    )
+    configuration = json.loads(result.stdout)
+    configured_services = configuration.get("services", {})
+    if any(service.get("build") is not None for service in configured_services.values()):
+        logging.info("Skipping project %s: contains locally built services", project.name)
+        return
+    services = []
+    for name in sorted(project.services):
+        service = configured_services.get(name)
+        if service is None:
+            raise ValueError(f"Project {project.name}: service {name} is absent from current Compose config")
+        image = service.get("image") or ""
+        if not image or is_pinned(image=image):
+            logging.warning("Skipping %s/%s: current Compose image %s is missing or pinned", project.name, name, image)
+        else:
+            logging.info("Refreshing %s/%s (%s)", project.name, name, image)
+            services.append(name)
+    if not services:
+        return
+    for arguments in (
+        ["pull", "--policy", "always", *services],
+        ["up", "--detach", "--no-deps", "--no-build", "--pull", "never", *services],
+    ):
+        logging.info("%s", shlex.join([*command, *arguments]))
+        if not dry_run:
+            subprocess.run(args=[*command, *arguments], cwd=project.directory, check=True)
+
+
+def main() -> int:
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
+    arguments = parse_args()
+    try:
+        selected_projects = set(arguments.projects)
+        projects = discover_projects(containers=inspect_containers(), selected_projects=selected_projects)
+        if not projects:
+            if selected_projects:
+                logging.info("No eligible services to refresh in selected projects: %s", ", ".join(sorted(selected_projects)))
+            else:
+                logging.info("No eligible services to refresh")
+        failed = False
+        for project in projects:
+            try:
+                refresh_project(project=project, dry_run=arguments.dry_run)
+            except (OSError, ValueError, subprocess.SubprocessError) as error:
+                logging.error("Project %s: %s", project.name, getattr(error, "stderr", None) or error)
+                failed = True
+        return int(failed)
+    except (OSError, ValueError, subprocess.SubprocessError) as error:
+        logging.error("%s", getattr(error, "stderr", None) or error)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())