Delete AMI's and related snapshots

purge_amis.py

import json

import boto3
from botocore.exceptions import ClientError


AMIS_FILE = "amis.json"
REFRESH_AMIS_FILE = False
DRY_RUN = True

ec2 = boto3.client('ec2')


def get_all_amis():
    resp = ec2.describe_images(Owners=['self'])
    with open(AMIS_FILE, "w") as f:
        json.dump(resp, f)


def get_ami_snapshots(ami):
    return [x["Ebs"]["SnapshotId"] for x in ami["BlockDeviceMappings"] if "Ebs" in x]


def delete_amis_by_prefix(prefix):
    for ami in get_amis_by_prefix(prefix):
        print(f"{ami['ImageId']} - {ami['Name']}")
        delete_ami(ami)
        snapshots = get_ami_snapshots(ami)
        for snapshot_id in snapshots:
            print(f"\tSnapshot {snapshot_id}")
            delete_snapshot(snapshot_id)
        print("")


def delete_ami(ami):
    if not DRY_RUN:
        try:
            ec2.deregister_image(ImageId=ami['ImageId'])
        except ClientError as ex:
            if ex.response["Error"]["Code"] != "InvalidAMIID.Unavailable":
                raise ex
            print("\tCould not delete AMI")


def delete_snapshot(snapshot_id):
    if not DRY_RUN:
        try:
            ec2.delete_snapshot(SnapshotId=snapshot_id)
        except ClientError as ex:
            if ex.response["Error"]["Code"] != "InvalidSnapshot.NotFound":
                raise ex
            print("\tCould not delete snapshot")


def get_amis_by_prefix(prefix):
    with open(AMIS_FILE) as f:
        amis = json.load(f)
    for image in amis["Images"]:
        if image["Name"].startswith(prefix):
            yield image


if __name__ == '__main__':
    if REFRESH_AMIS_FILE:
        get_all_amis()

    delete_amis_by_prefix("whatever-is-the-prefix")