release-cli multi-platform build script

.dockerignore

Dockerfile
.dockerignore
build.py

### .gitignore

# Ignore binaries, cover profiles and build files
*.out
.tmp
builds/
bin/
out/
gl-code-quality-report.json

# Ignore Visual Studio Code internals
/.vscode
/debug
debug.test

# Ignore Goland internals
.idea/

# Ignore the generated binary
/release-cli*

# Ignore vendor since we're using modules
vendor/

build.py

from __future__ import annotations
from argparse import ArgumentParser, BooleanOptionalAction
from contextlib import AbstractContextManager
from dataclasses import dataclass
from functools import reduce, wraps
from json import loads
from pathlib import Path
from pickle import dump, load
import re
from subprocess import PIPE, CalledProcessError, run
from types import TracebackType
from typing import Callable, Generic, Hashable, Iterable, Mapping, NotRequired, ParamSpec, Sequence, TypeVar, TypedDict, cast


WORKING_DIRECTORY = Path.cwd()
CACHE_DIR = WORKING_DIRECTORY / "out" / "cache"
FROM_REGEX = re.compile(r"^\s*FROM\s+(?:--platform=(\w+(?:\/\w+)*)\s+)?([\w.-]+(?::[\w.-]+|@\w+:\w+)?)(?:\s+AS\s+(\w+))?\s*$", flags=re.IGNORECASE)
BANNED_PLATFORMS = set([
    "linux/arm/v6",
])
T = TypeVar("T")
S = TypeVar("S", bound=Hashable)
Params = ParamSpec("Params")

class Platform(TypedDict):
    architecture: str
    os: str
    variant: NotRequired[str]

def get_platform_name(platform: Platform):
    name = f"{platform['os']}/{platform['architecture']}"
    if "variant" in platform:
        name = f"{name}/{platform['variant']}"
    return name

class ManifestsItem(TypedDict):
    mediaType: str
    size: int
    digest: str
    platform: Platform

class ManifestList(TypedDict):
    schemaVersion: int
    mediaType: str
    manifests: list[ManifestsItem]

@dataclass
class FromInstruction:
    image: str
    name: str | None = None
    platform: str | None = None

    @classmethod
    def from_match(cls, match: re.Match[str]):
        return cls(match.group(2), match.group(3), match.group(1))

def get_output(cmd: str | bytes | Sequence[str] | Sequence[bytes]):
    try:
        result = run(cmd, check=True, stdout=PIPE, stderr=PIPE)
    except CalledProcessError as e:
        if isinstance(e.stderr, bytes):
            raise Exception(e.stderr.decode().strip())
        raise Exception("Unknown error!")
    return result.stdout.decode()

def get_current_tag():
    tag = get_output(['git', 'describe', '--exact-match']).strip()
    prefix = "v"
    if not tag.startswith(prefix): raise Exception(f"Not a version tag {tag=}!")
    tag = tag[len(prefix):]
    return tag

def buildx(platforms: Iterable[str], tags: Iterable[str], build_args: Mapping[str, object] | None = None, no_cache: bool = False, builder: str | None = None, file: Path | None = None, dryrun: bool = False, push: bool = True, context: Path | None = None):
    context = context or WORKING_DIRECTORY
    cmd = ["docker", "buildx", "build"]
    if build_args is not None:
        for build_arg in build_args.items():
            cmd.extend(["--build-arg", f"{build_arg[0]}={str(build_arg[1])}"])
    if builder is not None:
        cmd.extend(["--builder", builder])
    if file is not None:
        cmd.extend(["--file", str(file)])
    if no_cache:
        cmd.append("--no-cache")
    for platform in platforms:
        cmd.extend(["--platform", platform])
    for tag in tags:
        cmd.extend(["--tag", tag])
    if push:
        cmd.append("--push")
    cmd.append(str(context))
    if dryrun:
        print(cmd)
        return
    run(cmd, check=True)

def parse_args():
    parser = ArgumentParser(description="Build multi-platform docker images")
    parser.add_argument("-t", "--tag", type=str, required=False, help="Tag to build, default acquired from current commit description")
    parser.add_argument("-r", "--repository", type=str, default="chenio/release-cli", help="Image repository to push towards")
    parser.add_argument("-p", "--push", action=BooleanOptionalAction, default=True, help="Whether to push to the remote repository")
    parser.add_argument("-f", "--file", type=Path, default=Path("Dockerfile"), help="Path of the Dockerfile")
    parser.add_argument("-l", "--latest", action=BooleanOptionalAction, default=True, help="Whether to also tag as latest")
    parser.add_argument("-d", "--dryrun", action=BooleanOptionalAction, default=False, help="Whether to not execute the build command")
    return parser.parse_args()

class CacheDb(AbstractContextManager["CacheDb[S, T]"], Generic[S, T]):
    db: dict[S, T] | None = None
    file: Path
    clean: bool = True

    def __init__(self, file: Path):
        self.file = file
        self.load_db()
        return super().__init__()

    def get(self, item: S):
        if self.db is None:
            raise Exception("Db was not yet loaded!")
        return self.db.get(item)

    def was_changed(self):
        self.clean = False
        self.write_changes()
    
    def __setitem__(self, item: S, value: T):
        if self.db is None:
            raise Exception("Db was not yet loaded!")
        self.db[item] = value
        self.was_changed()
    
    def __enter__(self) -> CacheDb[S, T]:
        self.load_db()
        return self

    def load_db(self):
        try:
            with self.file.open("rb") as f:
                self.db = cast(dict[S, T], load(f))
        except FileNotFoundError:
            self.db = dict[S, T]()
        return self.db

    def write_changes(self):
        if self.clean: return
        with self.file.open("wb") as f:
            dump(self.db, f)

    def __exit__(self, __exc_type: type[BaseException] | None, __exc_value: BaseException | None, __traceback: TracebackType | None) -> bool | None:
        self.write_changes()


def cached(f: Callable[[S], T]) -> Callable[[S], T]:
    cache = CacheDb[S, T](CACHE_DIR / f"{f.__module__}.{f.__name__}.cache")
    @wraps(f)
    def wrapped(item: S):
        if (result := cache.get(item)) is not None:
            return result
        result = f(item)
        cache[item] = result
        return result
    return wrapped

@cached
def get_manifest_list(image: str) -> ManifestList:
    return loads(get_output(["docker", "manifest", "inspect", image]))

def get_supported_platforms(image: str):
    manifest_list = get_manifest_list(image)
    return (manifest["platform"] for manifest in manifest_list["manifests"])

def as_set(items: Iterable[T]) -> set[T]:
    return items if isinstance(items, set) else set(items)

def set_intersect(a: Iterable[T], b: Iterable[T]):
    return as_set(a).intersection(b)

def determine_platforms(dockerfile: Path):
    with dockerfile.open() as f:
        lines = f.readlines()
    matches = map(FROM_REGEX.match, lines)
    matches = filter(None, matches)
    froms = map(FromInstruction.from_match, matches)
    no_platform = filter(lambda x: x.platform is None, froms)
    supported_platforms = (map(get_platform_name, get_supported_platforms(instr.image)) for instr in no_platform)
    platforms: set[str] = as_set(reduce(set_intersect, supported_platforms))
    platforms.difference_update(BANNED_PLATFORMS)
    return platforms
    
def get_repository_name():
    return WORKING_DIRECTORY.name

def main():
    args = parse_args()
    repository: str = args.repository
    # repository = repository or get_repository_name()
    dockerfile: Path = args.file
    tag: str | None = args.tag
    tag = tag or get_current_tag()
    tags = [tag]
    latest: bool = args.latest
    dryrun: bool = args.dryrun
    if latest:
        tags.append("latest")
    push: bool = args.push
    images = (f"{repository}:{tag}" for tag in tags)
    platforms = determine_platforms(dockerfile)
    buildx(platforms, images, push=push, dryrun=dryrun)

if __name__ == '__main__':
    main()