Skip to content

Instantly share code, notes, and snippets.

@sassyn
Forked from fabidick22/migrate-ecr.py
Created June 26, 2021 18:13
Show Gist options
  • Save sassyn/ebd2ea0beb8841cdd8e734cb772585bf to your computer and use it in GitHub Desktop.
Save sassyn/ebd2ea0beb8841cdd8e734cb772585bf to your computer and use it in GitHub Desktop.
Migrate AWS ECR images from one account to another in different regions
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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment