s3_operations.py

import os

import boto3
from botocore.exceptions import ClientError
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

class S3Operations:
    """Amazon S3 operations including bucket creation, file upload, and static website hosting"""
    
    def __init__(self):
        """Initialize S3 client with AWS credentials from environment variables """
        self.s3_client = boto3.client(
            's3',
            aws_access_key_id=os.getenv('AWS_ACCESS_KEY_ID'),
            aws_secret_access_key=os.getenv('AWS_SECRET_ACCESS_KEY'),
            region_name=os.getenv('AWS_REGION', 'eu-north-1')
        )

    def create_bucket(self, bucket_name):
        """
        Create a new S3 bucket.
        Args:
            bucket_name (str): Name of the bucket to create
        Returns:
            bool: True if successful, False if failed
        """
        try:
            self.s3_client.create_bucket(Bucket=bucket_name)
            print(f"Bucket {bucket_name} created successfully")
        except ClientError as e:
            print(f"Error creating bucket: {e}")
            return False
        return True

    def upload_file(self, file_path, bucket_name, object_name=None):
        """
        Upload a file to an S3 bucket.
        Args:
            file_path (str): Path to file to upload
            bucket_name (str): Bucket to upload to
            object_name (str): S3 object name (defaults to file name)
        Returns:
            bool: True if successful, False if failed
        """
        if object_name is None:
            object_name = os.path.basename(file_path)

        try:
            self.s3_client.upload_file(file_path, bucket_name, object_name)
            print(f"File {file_path} uploaded successfully to {bucket_name}")
        except ClientError as e:
            print(f"Error uploading file: {e}")
            return False
        return True

    def list_buckets(self):
        """
        List all S3 buckets in the AWS account.
        Prints bucket names to console.
        """
        try:
            response = self.s3_client.list_buckets()
            print("Existing buckets:")
            for bucket in response['Buckets']:
                print(f"- {bucket['Name']}")
        except ClientError as e:
            print(f"Error listing buckets: {e}")

    def enable_static_website(self, bucket_name, index_doc='index.html'):
        """
        Enable static website hosting for an S3 bucket.
        Args:
            bucket_name (str): Name of the bucket
            index_doc (str): Name of the index document (default: index.html)
        Returns:
            bool: True if successful, False if failed
        """
        try:
            website_configuration = {
                'IndexDocument': {'Suffix': index_doc}
            }
            self.s3_client.put_bucket_website(
                Bucket=bucket_name,
                WebsiteConfiguration=website_configuration
            )
            print(f"Static website hosting enabled for {bucket_name}")
        except ClientError as e:
            print(f"Error enabling static website: {e}")
            return False
        return True

def main():
    """Main function to demonstrate S3 operations."""
    s3_ops = S3Operations()
    # Example usage with a sample bucket name
    bucket_name = "polly-notes-bucket"
    # List existing buckets in the account
    s3_ops.list_buckets()
    # Create a new bucket for the website
    s3_ops.create_bucket(bucket_name)
    # Configure the bucket for static website hosting
    s3_ops.enable_static_website(bucket_name)

if __name__ == "__main__":
    main()