ftb-docker/docker.py

from subprocess import PIPE, run
from typing import Literal, NotRequired, TypedDict, get_args, List
from json import loads

Platforms = Literal["linux/amd64", "linux/amd64/v2", "linux/amd64/v3", "linux/arm64", "linux/riscv64", "linux/ppc64le", "linux/s390x", "linux/386", "linux/mips64le", "linux/mips64", "linux/arm/v7", "linux/arm/v6"]
supported_platforms: list[Platforms] = list(get_args(Platforms))

Progress = Literal["auto", "plain", "tty"]

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


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


class Manifest(TypedDict):
    schemaVersion: int
    mediaType: str
    digest: str
    size: int
    manifests: List[ManifestsItem]


class Config(TypedDict):
    Env: List[str]
    Cmd: List[str]


class Rootfs(TypedDict):
    type: str
    diff_ids: List[str]


class HistoryItem(TypedDict):
    created: str
    created_by: str
    empty_layer: NotRequired[bool]


class PlatformImage(TypedDict):
    created: str
    architecture: str
    variant: NotRequired[str]
    os: str
    config: Config
    rootfs: Rootfs
    history: List[HistoryItem]


class ImagetoolsInspectResult(TypedDict):
    name: str
    manifest: Manifest
    image: dict[Platform, PlatformImage]

def buildx_imagetools_inspect(repository: str, tag: str) -> ImagetoolsInspectResult:
    label = f"{repository}:{tag}"
    process = run(["docker", "buildx", "imagetools", "inspect", label, "--format", "{{json .}}"], check=True, stdout=PIPE)
    output = process.stdout.decode("utf-8")
    return loads(output)

def semver_tags(version: str):
    parts = version.split('.')
    return list('.'.join(parts[:i]) for i in range(1, len(parts) + 1))

def pull(repository: str, tag: str):
    label = f"{repository}:{tag}"
    run(["docker", "pull", label], check=True)
    return label

def push_image(repository: str, tag: str):
    label = f"{repository}:{tag}"
    run(["docker", "push", label], check=True)
    return label

def create_manifest(repository: str, manifest_tag: str, image_tags: list[str], push: bool = True, write_command: bool = False):
    raise Exception("Creation of manifests is not yet supported!")
    manifest = f"{repository}:{manifest_tag}"
    images = [f"{repository}:{tag}" for tag in image_tags]
    for image_tag in image_tags:
        pull(repository, image_tag)
    command = ["docker", "manifest", "create", manifest, *images]
    if write_command:
        print(" ".join(command))
    run(command, check=True)
    if push:
        return push_image(repository, manifest_tag)
    return manifest

def buildx(repository: str, tags: list[str], build_platforms: list[Platforms], dockerfile: str = "Dockerfile", build_args: dict[str, str] | None = None, directory: str = ".", push: bool = True, pull: bool = False, progress: Progress = "auto", write_command: bool = False):
    if build_args is None:
        build_args = dict()
    labels = [f"{repository}:{tag}" for tag in tags]
    command = list(filter(None, ["docker", "buildx", "build",
               "--platform", ",".join(build_platforms),
               *[t for (key, value) in build_args.items() for t in ("--build-arg", f"{key}={value}")],
               "--file", dockerfile,
               *[t for label in labels for t in ("--tag", label)],
               f"--progress={progress}",
               "--pull" if pull else None,
               "--push" if push else None,
               directory]))
    if write_command:
        print(" ".join(command))
    run(command, check=True)