#!/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())