How to Delete All S3 Buckets using python

Photo by Ilya Pavlov on Unsplash

How to Delete All S3 Buckets using python

S3 version Files along with Delete marker files using python SDK

ยท

2 min read

Before you run this script:

  1. Install Boto3 using pip if you haven't already: pip install boto3.

  2. Configure your AWS credentials properly. You can either set them up using the AWS CLI (aws configure) or set environment variables (AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY).

  3. Ensure that the IAM user associated with your credentials has the necessary permissions to delete S3 buckets and objects. Be cautious about granting such permissions.

Make sure check your I AM console does user contains privileges to delete the Resources. In My case I have provide Admin Privileges to my user.

Also, I have added this code in case you are unable to delete a bucket due to certain reasons.

PS C:\Users\je.balakrishnan> & C:/Users/je.balakrishnan/AppData/Local/Microsoft/WindowsApps/python3.10.exe "c:/Users/je.balakrishnan/OneDrive - Inc/Downloads/delete buckets -jeeva.py" Emptying bucket elasticbeanstalk-us-east-1-693527421032... Deleting bucket elasticbeanstalk-us-east-1-693527421032... Traceback (most recent call last): File "c:\Users\je.balakrishnan\OneDrive - Inc\Downloads\delete buckets -jeeva.py", line 47, in delete_all_buckets() File "c:\Users\je.balakrishnan\OneDrive - Inc\Downloads\delete buckets -jeeva.py", line 43, in delete_all_buckets s3.delete_bucket(Bucket=bucket_name) File "C:\Users\je.balakrishnan\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\botocore\client.py", line 530, in apicall return self._make_api_call(operation_name, kwargs) File "C:\Users\je.balakrishnan\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\botocore\client.py", line 960, in makeapi_call raise error_class(parsed_response, operation_name) botocore.exceptions.ClientError: An error occurred (AccessDenied) when calling the DeleteBucket operation: Access Denied

Code Reference:

import boto3
import botocore

def empty_bucket(s3_client, bucket_name):
    # List all objects including versions
    response = s3_client.list_object_versions(Bucket=bucket_name)

    # Delete all objects including versions
    if 'Versions' in response:
        for obj in response['Versions']:
            s3_client.delete_object(Bucket=bucket_name, Key=obj['Key'], VersionId=obj['VersionId'])
            print(f"Deleted versioned object: {obj['Key']}")

    # Delete all delete markers
    if 'DeleteMarkers' in response:
        for marker in response['DeleteMarkers']:
            s3_client.delete_object(Bucket=bucket_name, Key=marker['Key'], VersionId=marker['VersionId'])
            print(f"Deleted delete marker: {marker['Key']}")

    # Empty the bucket
    response = s3_client.list_objects_v2(Bucket=bucket_name)
    if 'Contents' in response:
        for obj in response['Contents']:
            s3_client.delete_object(Bucket=bucket_name, Key=obj['Key'])
            print(f"Deleted object: {obj['Key']}")

def delete_all_buckets():
    # Create an S3 client
    s3 = boto3.client('s3')

    # List all buckets
    response = s3.list_buckets()

    # Iterate over each bucket, empty it, then delete it
    for bucket in response['Buckets']:
        bucket_name = bucket['Name']
        print(f"Emptying bucket {bucket_name}...")

        # Empty the bucket
        try:
            empty_bucket(s3, bucket_name)
        except botocore.exceptions.ClientError as e:
            if e.response['Error']['Code'] == 'AccessDenied':
                print(f"Access denied. Skipping bucket deletion: {bucket_name}")
                continue
            else:
                raise

        print(f"Deleting bucket {bucket_name}...")
        # Delete the bucket
        try:
            s3.delete_bucket(Bucket=bucket_name)
            print(f"Deleted bucket: {bucket_name}")
        except botocore.exceptions.ClientError as e:
            if e.response['Error']['Code'] == 'AccessDenied':
                print(f"Access denied. Skipping bucket deletion: {bucket_name}")
            else:
                raise

if __name__ == "__main__":
    delete_all_buckets()
ย