expunge-bucket

#!/usr/bin/env python3
import boto3
import logging
from argparse import ArgumentParser
from logging import info, error
from operator import attrgetter, itemgetter
from sys import exit

argparser = ArgumentParser()
argparser.add_argument("bucket")
argparser.add_argument("--dry-run", "-n", action = "store_true")

bucket, dry_run = attrgetter("bucket", "dry_run")(argparser.parse_args())

loggingConfig = {
    "level": "INFO",
    "format": (
        f"%(asctime)s [DRY RUN] %(levelname)8s %(message)s"
            if dry_run else
        f"%(asctime)s %(levelname)8s %(message)s"
    ),
}
logging.basicConfig(**loggingConfig)

info(f"Expunging bucket s3://{bucket}")

exit_status = 0

s3 = boto3.client("s3")
list_object_versions = s3.get_paginator("list_object_versions")

for page in list_object_versions.paginate(Bucket = bucket, PaginationConfig={"PageSize":1000}):
    versions = [
        {"Key": version["Key"], "VersionId": version["VersionId"]}
        for version
         in [*page.get("Versions", []), *page.get("DeleteMarkers", [])]
    ]

    if not versions:
        continue

    info(f"Deleting {len(versions):,} object versions…")

    if dry_run:
        response = {"Deleted": versions, "Errors": []}
    else:
        response = s3.delete_objects(Bucket = bucket, Delete = {"Objects": versions})

    deleted = response.get("Deleted", [])
    errors  = response.get("Errors", [])

    for key, version_id in map(itemgetter("Key", "VersionId"), deleted):
        info(f"deleted: s3://{bucket}/{key}?VersionId={version_id}")

    if errors:
        exit_status = 1
        for key, version_id, msg, code in map(itemgetter("Key", "VersionId", "Message", "Code"), errors):
            error(f"{msg} ({code}): s3://{bucket}/{key}?VersionId={version_id}")

try:
    if not dry_run:
        s3.delete_bucket(Bucket = bucket)
except s3.exceptions.ClientError as e:
    error(f"{e}: s3://{bucket}")
else:
    info(f"deleted bucket: s3://{bucket}")

exit(exit_status)