CPP_30.9.25

# s3_demo.py
# Interactive S3 management tool with menu
# Requires boto3

import logging
import boto3
from botocore.exceptions import ClientError


def create_bucket(bucket_name, region=None):
    """Create an S3 bucket in a specified region"""
    try:
        if region is None:
            s3_client = boto3.client('s3')
            s3_client.create_bucket(Bucket=bucket_name)
        else:
            s3_client = boto3.client('s3', region_name=region)
            location = {'LocationConstraint': region}
            s3_client.create_bucket(Bucket=bucket_name,
                                    CreateBucketConfiguration=location)
        print(f"Bucket created: {bucket_name}")

    except ClientError as e:
        error_code = e.response['Error']['Code']
        if error_code == 'BucketAlreadyOwnedByYou':
            print(f"Bucket '{bucket_name}' already exists and is owned by you.")
        elif error_code == 'BucketAlreadyExists':
            print(f"Bucket '{bucket_name}' already exists. Bucket names must be globally unique. Try a different name.")
        else:
            logging.error(e)
            print("Error creating bucket.")


def list_buckets():
    """List all buckets"""
    s3_client = boto3.client('s3')
    response = s3_client.list_buckets()
    if not response['Buckets']:
        print("No buckets found.")
    else:
        print("Existing buckets:")
        for bucket in response['Buckets']:
            print("\t", bucket["Name"])


def upload_file(file_name, bucket, object_key=None):
    """Upload a file to an S3 bucket"""
    if object_key is None:
        object_key = file_name

    s3_client = boto3.client('s3')
    try:
        s3_client.upload_file(file_name, bucket, object_key)
        print(f"Uploaded {file_name} to {bucket}/{object_key}")
    except ClientError as e:
        logging.error(e)
        print("Error uploading file.")


def list_objects(bucket_name):
    """List all objects in a specific S3 bucket"""
    s3_client = boto3.client('s3')
    try:
        response = s3_client.list_objects_v2(Bucket=bucket_name)
        if response['KeyCount'] == 0:
            print(f"No objects found in bucket '{bucket_name}'.")
        else:
            print(f"Objects in bucket '{bucket_name}':")
            for obj in response['Contents']:
                print("\t", obj['Key'])
    except ClientError as e:
        logging.error(e)
        print(f"Error listing objects in bucket '{bucket_name}'.")


def delete_object(bucket_name, object_key):
    """Delete an object from an S3 bucket"""
    s3_client = boto3.client('s3')
    try:
        s3_client.delete_object(Bucket=bucket_name, Key=object_key)
        print(f"Deleted object {object_key} from {bucket_name}")
    except ClientError as e:
        logging.error(e)
        print(f"Error deleting object {object_key} from {bucket_name}")


def delete_bucket(bucket_name):
    """Delete a bucket and optionally its contents"""
    s3_client = boto3.client('s3')
    try:
        response = s3_client.list_objects_v2(Bucket=bucket_name)
        if response['KeyCount'] != 0:
            print(f"Bucket '{bucket_name}' is not empty. It contains {response['KeyCount']} objects.")
            choice = input("Do you want to delete all objects and the bucket? (y/n): ").lower()
            if choice == 'y':
                for content in response['Contents']:
                    object_key = content['Key']
                    print(f"Deleting object {object_key} ...")
                    s3_client.delete_object(Bucket=bucket_name, Key=object_key)
            else:
                print("Bucket deletion cancelled.")
                return

        s3_client.delete_bucket(Bucket=bucket_name)
        print(f"Deleted bucket {bucket_name}")
    except ClientError as e:
        logging.error(e)
        print(f"Error deleting bucket {bucket_name}")

def download_file(bucket_name, object_key, file_name=None):
    """Download an S3 object to a local file"""
    if file_name is None:
        file_name = object_key

    s3_client = boto3.client('s3')
    try:
        s3_client.download_file(bucket_name, object_key, file_name)
        print(f"Downloaded {bucket_name}/{object_key} to {file_name}")
    except ClientError as e:
        logging.error(e)
        print(f"Error downloading {object_key} from {bucket_name}")


def enable_versioning(bucket_name):
    """Enable versioning on an S3 bucket and confirm status"""
    s3_client = boto3.client('s3')
    try:
        # Enable versioning
        s3_client.put_bucket_versioning(
            Bucket=bucket_name,
            VersioningConfiguration={'Status': 'Enabled'}
        )

        # Check versioning status
        response = s3_client.get_bucket_versioning(Bucket=bucket_name)
        status = response.get('Status', 'Disabled')
        print(f"Versioning status for bucket '{bucket_name}': {status}")

        if status == 'Enabled':
            print(f"Versioning successfully enabled on bucket '{bucket_name}'")
        else:
            print(f"Failed to enable versioning on bucket '{bucket_name}'")

    except ClientError as e:
        logging.error(e)
        print(f"Error enabling versioning on bucket '{bucket_name}'")





def menu():
    while True:
        print("\nS3 Demo Menu")
        print("1. Create bucket")
        print("2. List buckets")
        print("3. Upload file")
        print("4. List objects in a bucket")
        print("5. Delete object")
        print("6. Delete bucket")
        print("7. Download object")
        print("8. Enable bucket versioning")
        print("9. Exit")

        choice = input("Enter your choice (1-9): ")

        if choice == "1":
            bucket_name = input("Enter bucket name: ")
            region = input("Enter region (leave blank for default us-east-1): ") or None
            create_bucket(bucket_name, region)

        elif choice == "2":
            list_buckets()

        elif choice == "3":
            bucket_name = input("Enter bucket name: ")
            file_name = input("Enter file name to upload: ")
            object_key = input("Enter object key (leave blank to use file name): ") or None
            upload_file(file_name, bucket_name, object_key)

        elif choice == "4":
            bucket_name = input("Enter bucket name to list objects: ")
            list_objects(bucket_name)

        elif choice == "5":
            bucket_name = input("Enter bucket name: ")
            object_key = input("Enter object key to delete: ")
            delete_object(bucket_name, object_key)

        elif choice == "6":
            bucket_name = input("Enter bucket name to delete: ")
            delete_bucket(bucket_name)

        elif choice == "7":
            bucket_name = input("Enter bucket name: ")
            object_key = input("Enter object key to download: ")
            file_name = input("Enter local file name (leave blank to use object key): ") or None
            download_file(bucket_name, object_key, file_name)

        elif choice == "8":
            bucket_name = input("Enter bucket name to enable versioning: ")
            enable_versioning(bucket_name)

        elif choice == "9":
            print("Exiting...")
            break

        else:
            print("Invalid choice. Please try again.")


if __name__ == "__main__":
    menu()