AWS S3 Cross Account Copy Files

import boto3
from botocore.exceptions import ClientError
import logging
import sys
import argparse

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def create_s3_client(profile_name, region):
    """Create an S3 client with specified profile and region"""
    try:
        session = boto3.Session(profile_name=profile_name)
        return session.client('s3', region_name=region)
    except Exception as e:
        logger.error(f"Error creating S3 client: {str(e)}")
        raise

def copy_objects(source_client, dest_client, source_bucket, dest_bucket, prefix=''):
    """Copy objects from source bucket to destination bucket"""
    try:
        # List objects in source bucket
        paginator = source_client.get_paginator('list_objects_v2')
        
        for page in paginator.paginate(Bucket=source_bucket, Prefix=prefix):
            if 'Contents' not in page:
                continue
                
            for obj in page['Contents']:
                source_key = obj['Key']
                
                # Create copy source dictionary
                copy_source = {
                    'Bucket': source_bucket,
                    'Key': source_key
                }
                
                logger.info(f"Copying {source_key} to {dest_bucket}")
                
                try:
                    # Perform the copy operation
                    dest_client.copy_object(
                        CopySource=copy_source,
                        Bucket=dest_bucket,
                        Key=source_key
                    )
                except ClientError as e:
                    logger.error(f"Error copying object {source_key}: {str(e)}")
                    continue
                
    except ClientError as e:
        logger.error(f"Error listing objects: {str(e)}")
        raise

def parse_arguments():
    """Parse command line arguments"""
    parser = argparse.ArgumentParser(description='Copy files between S3 buckets across AWS accounts')
    parser.add_argument('--source-bucket', required=True, help='Source S3 bucket name')
    parser.add_argument('--dest-bucket', required=True, help='Destination S3 bucket name')
    parser.add_argument('--prefix', default='', help='Optional prefix/path to copy specific files')
    parser.add_argument('--source-profile', default='profile_a', help='AWS profile for source account')
    parser.add_argument('--dest-profile', default='profile_b', help='AWS profile for destination account')
    parser.add_argument('--source-region', default='us-east-1', help='Region for source bucket')
    parser.add_argument('--dest-region', default='us-east-1', help='Region for destination bucket')
    return parser.parse_args()

def main():
    # Parse command line arguments
    args = parse_arguments()

    try:
        # Create S3 clients for both accounts
        source_client = create_s3_client(args.source_profile, args.source_region)
        dest_client = create_s3_client(args.dest_profile, args.dest_region)

        # Copy objects
        copy_objects(
            source_client, 
            dest_client, 
            args.source_bucket, 
            args.dest_bucket,
            args.prefix
        )
        logger.info("Copy operation completed successfully")

    except Exception as e:
        logger.error(f"An error occurred: {str(e)}")
        sys.exit(1)

if __name__ == '__main__':
    """
    USAGE:
    python copy_s3_btw_accounts.py \
        --source-bucket my-source-bucket \
        --dest-bucket my-dest-bucket \
        --prefix my-folder/ \
        --source-profile profile_a \
        --dest-profile profile_b \
        --source-region us-east-1 \
        --dest-region us-west-2
    """
    main()