ecr-clean.py

from __future__ import print_function
import sys
import boto3
import logging
from botocore.exceptions import ParamValidationError

ecr = boto3.client('ecr')

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def delete_images(repository, images=[]):
    logger.info('Deleting {} images from {}'.format(len(images), repository))
    if images:
        try:
            response = ecr.batch_delete_image(
                repositoryName=repository,
                imageIds=[]
            )
            deleted_images = response['imageIds']
            failure_images = response['failures']
            logger.info('Successfully deleted {}'.format(len(deleted_images)))
            logger.warn('{} could not be deleted'.format(len(failure_images)))

            for image in failure_images:
                image_id = image['imageId']['imageDigest']
                logger.warn('Failed to delete image {}'.format(image_id))
                logger.warn('Caused by {} {}'.format(
                    image['failureCode'], image['failureReason']))
        except ParamValidationError as e:
            print(e)
            raise e


def find_untagged_images(repository):
    logger.info('Scanning untagged images from {}'.format(repository))
    response = ecr.list_images(
        repositoryName=repository,
        filter={'tagStatus': 'UNTAGGED'}
    )
    return response['imageIds']


if __name__ == '__main__':
    repositories = sys.argv[1:]
    for repository in repositories:
        images = find_untagged_images(repository)
        delete_images(repository, images)