AWS ECR tag multi-arch images

main.py

import typer
import json
import subprocess

IMAGE_INDEX_IMAGE_MANIFEST_MEDIA_TYPE="application/vnd.docker.distribution.manifest.list.v2+json"
DEFAULT_NEW_TAG_TEMPLATE="{image_index_tag}-{platform_os}-{platform_arch}"
GET_MANIFEST_COMMAND_TEMPLATE="MANIFEST=$(aws ecr batch-get-image --repository-name {repo_name} --image-ids imageDigest='{digest}' --output json | jq --raw-output --join-output '.images[0].imageManifest')"
PUT_IMAGE_COMMAND_TEMPLATE='aws ecr put-image --repository-name {repo_name} --image-tag {tag} --image-manifest \"$MANIFEST\"'

def get_tag_from_image(image):
    """
        Filters non-"latest" tags from image tag list
    """
    image_tags = image['imageTags']
    filtered_non_latest_tags = filter(lambda x: x != "latest", image_tags)
    return list(filtered_non_latest_tags)[0]

def get_image_manifest(repository_name, digest):
    """
        Gets image manifest from a given repository name and image digest hash
    """
    result = subprocess.run(["aws", "ecr", "batch-get-image", "--repository-name", repository_name, "--image-ids", "imageDigest=" + digest], stdout=subprocess.PIPE)
    dict = json.loads(result.stdout)
    return dict['images'][0]['imageManifest']

def process_image(image, image_index_tag, repository_name):
    """
        Adds a new tag to said image depending on image index tag and architecture
    """
    digest = image['digest']
    tag = image_index_tag
    os = image['platform']['os']
    arch = image['platform']['architecture']

    new_tag = DEFAULT_NEW_TAG_TEMPLATE.format(image_index_tag=tag, platform_os=os, platform_arch=arch)

    typer.echo(f"Processing image: {image['digest']}")

    try:
        result = subprocess.run(GET_MANIFEST_COMMAND_TEMPLATE.format(repo_name=repository_name, digest=digest) + ';' + PUT_IMAGE_COMMAND_TEMPLATE.format(repo_name=repository_name, tag=new_tag), shell=True, stdout=subprocess.PIPE)

        typer.echo(f"Created: {new_tag}")
    except subprocess.CalledProcessError as e:
        typer.echo(f"Error in: {new_tag}")
        typer.echo(e.output)

    return 0

def get_image_index_images(repository_name):
    """
        Filters image index type of images from said repository
    """
    result = subprocess.run(["aws", "ecr", "describe-images", "--repository-name", repository_name], stdout=subprocess.PIPE)
    dict = json.loads(result.stdout)

    image_list = dict['imageDetails']
    filtered = filter(lambda x: x['imageManifestMediaType'] == IMAGE_INDEX_IMAGE_MANIFEST_MEDIA_TYPE, image_list)
    filtered_list = list(filtered)
    return filtered_list
    
def process_image_index_list(image_index_list, repository_name):
    """
        For each image index, gets its manifest and processes its "sub-images"
    """
    for i in image_index_list:
        typer.echo(f"Processing image index: {i['imageDigest']}")

        image_manifest = json.loads(get_image_manifest(repository_name, i['imageDigest']))
        image_manifest_manifests_list = image_manifest['manifests']

        for j in image_manifest_manifests_list:
            process_image(j, get_tag_from_image(i), repository_name)

def main(repository_name: str):
    typer.echo(f"Processing {repository_name}")
    list = get_image_index_images(repository_name)
    process_image_index_list(list, repository_name)

if __name__ == "__main__":
    typer.run(main)