skaffold_fetch.py

from itertools import cycle
from multiprocessing.pool import ThreadPool
from sys import stdout
from subprocess import run
from time import sleep

SPINNER = cycle(["⠁", "⠂", "⠄", "⡀", "⢀", "⠠", "⠐", "⠈"])
DOCKER_IMAGES = ("arlyon/workout-server", "arlyon/workout-learn")


def main():
    hash_proc = run(["git", "log", "--pretty=format:%h", "-n 1"], capture_output=True)
    git_hash = hash_proc.stdout.decode()

    results = {}
    pool = ThreadPool(processes=len(DOCKER_IMAGES))
    for image in DOCKER_IMAGES:
        t = pool.apply_async(run,
                             args=(["docker", "pull", f"{image}:{git_hash}"],),
                             kwds={"capture_output": True}
                             )
        results[f"fetching {image}:{git_hash}"] = t
        print("")  # the loop below moves up a line for every thread

    while True:
        symbol = next(SPINNER)

        for _ in results:
            stdout.write("\033[F")

        for name, result in results.items():
            print_status(symbol, name, result)

        if all(result.ready() for result in results.values()):
            break

        sleep(0.1)


def print_status(symbol, name, result):
    if result.ready():
        result = result.get()
        if result.returncode == 0:
            message = "success"
            text = result.stdout.decode()
            if "up to date" in text:
                message = "found locally"
            elif "newer image" in text:
                message = "downloaded image"
            print(f"\u001b[32m✔ {name} - {message}\u001b[0m")
        else:
            print(f"\u001b[31m✗ {name} - image not available\u001b[0m")
    else:
        print(f"{symbol} {name}")


if __name__ == '__main__':
    main()