Delete S3 bucket object from specific date

cutoff-objects.py

#!/usr/bin/env python3
"""
S3 Parallel Bulk Delete Script
Deletes objects from S3 bucket before a specified date using parallel processing
"""

import boto3
import argparse
from datetime import datetime, timezone
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict
import sys
from botocore.exceptions import ClientError
import logging

# Logging setup
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)


class S3BulkDeleter:
    def __init__(
        self,
        bucket_name: str,
        prefix: str = "",
        cutoff_date: str = None,
        batch_size: int = 1000,
        max_workers: int = 10,
    ):
        """
        Initialize S3 Bulk Deleter

        Args:
            bucket_name: S3 bucket name
            prefix: Object key prefix to filter
            cutoff_date: Delete objects before this date (YYYY-MM-DD)
            batch_size: Number of objects per delete request (max 1000)
            max_workers: Number of parallel threads
        """
        self.bucket_name = bucket_name
        self.prefix = prefix
        self.batch_size = min(batch_size, 1000)  # AWS limit
        self.max_workers = max_workers

        # Parse cutoff date
        if cutoff_date:
            self.cutoff_date = datetime.strptime(cutoff_date, "%Y-%m-%d").replace(
                tzinfo=timezone.utc
            )
        else:
            self.cutoff_date = None

        # Initialize S3 client
        self.s3_client = boto3.client("s3")

        # Statistics
        self.total_objects = 0
        self.deleted_objects = 0
        self.failed_objects = 0

    def list_objects_to_delete(self) -> List[Dict]:
        """
        List all objects matching criteria

        Returns:
            List of object dictionaries with Key and LastModified
        """
        objects_to_delete = []
        continuation_token = None

        logger.info(f"Scanning bucket: {self.bucket_name}")
        logger.info(f"Prefix: {self.prefix or '(root)'}")
        logger.info(
            f"Cutoff date: {self.cutoff_date.strftime('%Y-%m-%d') if self.cutoff_date else 'None (all objects)'}"
        )

        while True:
            try:
                # Build list_objects_v2 parameters
                params = {
                    "Bucket": self.bucket_name,
                    "Prefix": self.prefix,
                    "MaxKeys": 1000,
                }

                if continuation_token:
                    params["ContinuationToken"] = continuation_token

                response = self.s3_client.list_objects_v2(**params)

                if "Contents" not in response:
                    break

                # Filter by date if specified
                for obj in response["Contents"]:
                    if (
                        self.cutoff_date is None
                        or obj["LastModified"] < self.cutoff_date
                    ):
                        objects_to_delete.append(
                            {"Key": obj["Key"], "LastModified": obj["LastModified"]}
                        )

                logger.info(
                    f"Scanned: {len(objects_to_delete)} objects found so far..."
                )

                # Check for more objects
                if not response.get("IsTruncated", False):
                    break

                continuation_token = response.get("NextContinuationToken")

            except ClientError as e:
                logger.error(f"Error listing objects: {e}")
                raise

        self.total_objects = len(objects_to_delete)
        logger.info(f"Total objects to delete: {self.total_objects}")

        return objects_to_delete

    def create_batches(self, objects: List[Dict]) -> List[List[Dict]]:
        """
        Split objects into batches for deletion

        Args:
            objects: List of object dictionaries

        Returns:
            List of batches, each batch is a list of objects
        """
        batches = []
        for i in range(0, len(objects), self.batch_size):
            batch = objects[i : i + self.batch_size]
            batches.append(batch)

        logger.info(
            f"Created {len(batches)} batches of up to {self.batch_size} objects each"
        )
        return batches

    def delete_batch(self, batch: List[Dict], batch_num: int) -> Dict:
        """
        Delete a single batch of objects

        Args:
            batch: List of objects to delete
            batch_num: Batch number for logging

        Returns:
            Dictionary with deletion results
        """
        try:
            # Prepare delete request
            delete_objects = {
                "Objects": [{"Key": obj["Key"]} for obj in batch],
                "Quiet": False,  # Return list of deleted objects
            }

            response = self.s3_client.delete_objects(
                Bucket=self.bucket_name, Delete=delete_objects
            )

            deleted_count = len(response.get("Deleted", []))
            errors = response.get("Errors", [])
            error_count = len(errors)

            if errors:
                for error in errors:
                    logger.error(f"Failed to delete {error['Key']}: {error['Message']}")

            logger.info(
                f"Batch {batch_num}: Deleted {deleted_count}, Failed {error_count}"
            )

            return {
                "deleted": deleted_count,
                "failed": error_count,
                "batch_num": batch_num,
            }

        except ClientError as e:
            logger.error(f"Batch {batch_num} failed: {e}")
            return {"deleted": 0, "failed": len(batch), "batch_num": batch_num}

    def delete_parallel(self, batches: List[List[Dict]]) -> None:
        """
        Delete batches in parallel using ThreadPoolExecutor

        Args:
            batches: List of object batches to delete
        """
        logger.info(f"Starting parallel deletion with {self.max_workers} workers...")

        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # Submit all batches
            future_to_batch = {
                executor.submit(self.delete_batch, batch, idx): idx
                for idx, batch in enumerate(batches, 1)
            }

            # Process completed batches
            for future in as_completed(future_to_batch):
                try:
                    result = future.result()
                    self.deleted_objects += result["deleted"]
                    self.failed_objects += result["failed"]

                    # Progress update
                    progress = (
                        (self.deleted_objects + self.failed_objects)
                        / self.total_objects
                        * 100
                    )
                    logger.info(
                        f"Progress: {progress:.1f}% ({self.deleted_objects}/{self.total_objects} deleted)"
                    )

                except Exception as e:
                    logger.error(f"Batch processing error: {e}")

    def run(self, dry_run: bool = False) -> None:
        """
        Main execution method

        Args:
            dry_run: If True, only list objects without deleting
        """
        try:
            # List objects
            objects = self.list_objects_to_delete()

            if not objects:
                logger.info("No objects found matching criteria")
                return

            # Show sample objects
            logger.info("\nSample objects to delete (first 5):")
            for obj in objects[:5]:
                logger.info(f"  - {obj['Key']} (Modified: {obj['LastModified']})")

            if dry_run:
                logger.info(f"\nDRY RUN: Would delete {len(objects)} objects")
                return

            # Confirm deletion
            print(f"\n{'='*60}")
            print(f"About to delete {len(objects)} objects from {self.bucket_name}")
            print(f"{'='*60}")
            confirm = input("Type 'DELETE' to confirm: ")

            if confirm != "DELETE":
                logger.info("Deletion cancelled")
                return

            # Create batches and delete
            batches = self.create_batches(objects)
            self.delete_parallel(batches)

            # Final summary
            print(f"\n{'='*60}")
            print(f"DELETION SUMMARY")
            print(f"{'='*60}")
            print(f"Total objects: {self.total_objects}")
            print(f"Successfully deleted: {self.deleted_objects}")
            print(f"Failed: {self.failed_objects}")
            print(f"Success rate: {(self.deleted_objects/self.total_objects*100):.1f}%")
            print(f"{'='*60}")

        except KeyboardInterrupt:
            logger.warning("\nOperation cancelled by user")
            sys.exit(1)
        except Exception as e:
            logger.error(f"Fatal error: {e}")
            sys.exit(1)


def main():
    parser = argparse.ArgumentParser(
        description="Delete S3 objects in parallel before a specified date",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Delete all objects before 2024-12-01
  %(prog)s my-bucket --cutoff-date 2024-12-01
  
  # Delete objects with prefix before date
  %(prog)s my-bucket --prefix logs/ --cutoff-date 2024-11-01
  
  # Dry run (list only, don't delete)
  %(prog)s my-bucket --cutoff-date 2024-12-01 --dry-run
  
  # Custom parallelization
  %(prog)s my-bucket --cutoff-date 2024-12-01 --workers 20 --batch-size 500
        """,
    )

    parser.add_argument("bucket", help="S3 bucket name")
    parser.add_argument("--prefix", default="", help="Object key prefix to filter")
    parser.add_argument(
        "--cutoff-date", help="Delete objects before this date (YYYY-MM-DD)"
    )
    parser.add_argument(
        "--batch-size",
        type=int,
        default=1000,
        help="Objects per delete request (max 1000, default: 1000)",
    )
    parser.add_argument(
        "--workers",
        type=int,
        default=10,
        help="Number of parallel workers (default: 10)",
    )
    parser.add_argument(
        "--dry-run", action="store_true", help="List objects without deleting"
    )
    parser.add_argument("--profile", help="AWS profile name")
    parser.add_argument("--region", help="AWS region")

    args = parser.parse_args()

    # Set AWS profile and region if specified
    if args.profile:
        boto3.setup_default_session(profile_name=args.profile)
    if args.region:
        boto3.setup_default_session(region_name=args.region)

    # Create deleter and run
    deleter = S3BulkDeleter(
        bucket_name=args.bucket,
        prefix=args.prefix,
        cutoff_date=args.cutoff_date,
        batch_size=args.batch_size,
        max_workers=args.workers,
    )

    deleter.run(dry_run=args.dry_run)


if __name__ == "__main__":
    main()

python3 cutoff-objects.py sample-dev-bucket --cutoff-date 2025-12-20