#!/bin/env python3 import subprocess from multiprocessing import Process from enum import Enum from time import sleep import yaml from argparse import ArgumentParser, BooleanOptionalAction class Action(Enum): PUSH = "--push" LOAD = "--load" def from_str(text: str) -> Enum: match text: case "push": return Action.PUSH case "load": return Action.LOAD class Platform(Enum): amd64 = "linux/amd64" arm64 = "linux/arm64" armv6 = "linux/arm/v6" armv7 = "linux/arm/v7" def from_str(text: str) -> Enum: match text: case "armv7": return Platform.armv7 case "arm64": return Platform.arm64 case "amd64": return Platform.amd64 case _: raise Exception(f"unknown platform: {text}") class PlatformConfig: def __init__(self, platform: Platform, dockerfile: str, directory: str, args: dict[str, str], tags: list[str]): self.platform = platform self.directory = directory self.dockerfile = dockerfile self.args = args self.tags = tags def run(self, registry: str, library: str, name:str, action: Action, repo: str, verbose: bool) -> int: command = ["docker", "buildx", "build", action.value, "--platform", self.platform.value, "-f", self.dockerfile] for tag in self.tags: command.append("--tag") command.append(f"{registry}/{library}/{name}:{tag}") for arg in self.args.items(): command.append("--build-arg") command.append(f"{arg[0]}={arg[1]}") command.append(self.directory) return subprocess.run(command, cwd=repo, stderr=(subprocess.DEVNULL if not verbose else None)).returncode def __str__(self) -> str: return f"PlatformConfig{{{self.platform}, {self.directory}, {self.dockerfile}, {self.args}, {self.tags}}}" def __repr__(self) -> str: return str(self) class Builder: def __init__(self, name: str, repo: str, platforms: list[PlatformConfig], action: Action, registry: str, library: str): self.name = name self.repo = repo self.platforms = platforms self.action = action self.registry = registry self.library = library def run(self, parallel: bool = False, verbose: bool = False): if parallel: try: processes = list[tuple[str,Process]]() for platform in self.platforms: print(f"Building {self.name} for {platform.platform.name}...") p = Process(target=platform.run, args=(self.registry, self.library, self.name, self.action, self.repo, verbose)) p.start() processes.append((platform.platform.name,p)) while len(processes): for p in processes: if not p[1].is_alive(): if p[1].exitcode == 0: processes.remove(p) print(f"Built {self.name} for {p[0]}.") else: raise Exception(f"failed to complete {self.name} for {p[0]}") sleep(0.1) except KeyboardInterrupt: for p in processes: p[1].terminate() print("ancelled") exit() else: for platform in self.platforms: print(f"Building {self.name} for {platform.platform.name}...") if platform.run(self.registry, self.library, self.name, self.action, self.repo, verbose) != 0: raise Exception(f"failed to complete build for {platform.platform.name}") print("Done!") def tag(self, tag: str) -> int: command = ["docker", "buildx", "imagetools", "create", "-t", f"{self.registry}/{self.library}/{self.name}:{tag}"] for platform in self.platforms: command.append(f"{self.registry}/{self.library}/{self.name}:{platform.tags[0]}") return subprocess.run(command).returncode def latest(self) -> int: return self.tag("latest") def __str__(self) -> str: return f"Builder{{{self.name}, {self.repo}, {self.platforms}, {self.action.name}, {self.registry}, {self.library}}}" def __repr__(self) -> str: return str(self) class Tag: def __init__(self, registry: str, library: str, name: str, tag: str, builds: list[str]): self.registry = registry self.library = library self.name = name self.tag_name = tag self.builds = builds def tag(self) -> int: command = ["docker", "buildx", "imagetools", "create", "-t", f"{self.registry}/{self.library}/{self.name}:{self.tag_name}"] for build in self.builds: command.append(f"{self.registry}/{self.library}/{self.name}:{build}") return subprocess.run(command).returncode def __str__(self) -> str: return f"Tag{{{self.registry}/{self.library}/{self.name}:{self.tag_name}, {self.builds}}}" def __repr__(self) -> str: return str(self) def parse_yaml(path: str) -> tuple[list[Builder], list[Tag]]: file = {} with open(path, "r") as f: file = yaml.safe_load(f) registry = file.get("registry", "docker.io") repos = file.get("repos", []) builders = list[Builder]() tags = list[Tag]() for repo in repos: name = repo["name"] library = repo.get("library", "library") repo_dir = repo.get("repo", "build") action = repo.get("action", "push") repo_args = repo.get("args", []) repo_tags = repo.get("tags", []) builds = repo.get("builds", []) platforms = list[PlatformConfig]() for build in builds: arch = build.get("arch", "amd64") dockerfile = build.get("dockerfile", "Dockerfile") context = build.get("context", ".") args = build.get("args", []) build_tags = build.get("tags", []) platforms.append(PlatformConfig(Platform.from_str(arch), dockerfile, context, { arg["key"]: arg["value"] for arg in repo_args + args }, build_tags)) builders.append(Builder(name, repo_dir, platforms, Action.from_str(action), registry, library)) for tag in repo_tags: tags.append(Tag(registry, library, name, tag["name"], tag["builds"])) return (builders, tags) if __name__ == "__main__": parser = ArgumentParser( prog="BuildxManager", description="configure container builds with yaml" ) parser.add_argument("-f", action="store", default="build.yml", dest="file") parser.add_argument("-p", action=BooleanOptionalAction, dest="parallel") parser.add_argument("-v", action=BooleanOptionalAction, dest="verbose") parser.add_argument("--dump-only", action=BooleanOptionalAction, dest="dump") args = parser.parse_args() builders, tags = parse_yaml(args.file) if args.dump: print(builders, tags, sep="\n") else: print(f"Performing jobs listed in {args.file}") for builder in builders: builder.run(parallel=args.parallel, verbose=args.verbose) for tag in tags: tag.tag()