Migrate AWS ECR images from one account to another in different regions

migrate-ecr.py

import docker
import sys
import boto3 as b3

ECR_URI = "{}.dkr.ecr.{}.amazonaws.com/{}"
TAG_URI = "{}:{}"


class Repository:
    def __init__(self, repository_name: str, region: str, account_id: str, specific_tag=None, profile="default"):
        self.repository_name = repository_name
        self.region = region
        self.account_id = account_id
        self.specific_tag = specific_tag
        self.profile = profile

    def get_uri(self) -> str:
        base = ECR_URI.format(self.account_id, self.region, self.repository_name)
        return TAG_URI.format(base, self.specific_tag) if self.specific_tag else base


class MigrationECR:
    def __init__(self, ecr_from: Repository = None, ecr_to: Repository = None):
        self.ecr_from = ecr_from
        self.ecr_to = ecr_to
        self.client = docker.from_env()
        self.ta = docker.APIClient

    @staticmethod
    def get_images(ecr: Repository, limit=123) -> list:
        b3.setup_default_session(profile_name=ecr.profile)
        cli = b3.client("ecr", region_name=ecr.region)
        response = cli.list_images(repositoryName=ecr.repository_name, maxResults=limit)
        if ecr.specific_tag:
            return list(filter(lambda img: img.get("imageTag", None) == ecr.specific_tag, response.get("imageIds", None)))
        return response.get("imageIds", None)

    def push_images(self, ecr: Repository):
        if ecr.specific_tag:
            b3.setup_default_session(profile_name=ecr.profile)
            self.client.images.push(ecr.get_uri())
        else:
            raise Exception("Tag not found! ({})".format(ecr.specific_tag))

    def pull_and_tag(self):
        try:
            b3.setup_default_session(profile_name=self.ecr_from.profile)
            image_from = self.client.images.pull(self.ecr_from.get_uri())
            image_from.tag(self.ecr_to.get_uri())
        except docker.errors.APIError as e:
            print("""Error: {} \n\nMake sure you are authenticated with AWS ECR: \n$(aws ecr get-login --region REGIN --no-include-email --profile PROFILE)
            """.format(e))
            sys.exit(1)

    def migrate(self):
        if self.ecr_from and self.ecr_to:
            list_images_from = self.get_images(self.ecr_from)
            for image in list_images_from:
                print("image to pushed: {}\n".format(image))
                self.ecr_from.specific_tag = image.get("imageTag")
                self.ecr_to.specific_tag = image.get("imageTag")
                self.pull_and_tag()
                self.push_images(self.ecr_to)
        else:
            raise Exception("Missing variables: {} or {}".format("ecr_from", "ecr_to"))


# for test
r1_from = Repository("ecr_repo_name", "us-west-2", "629xxxxxxxx", specific_tag="v1.1", profile="ccc-prod")
r2_to = Repository("ecr_repo_name", "us-west-1", "804xxxxxxxxx", specific_tag="v1.1", profile="ccc-uat")
migration = MigrationECR(r1_from, r2_to)
migration.migrate()