recursive uploading of directory to Amazon s3 with automatic MIME type detection (for static sites)

s3_upload.py

import os
import sys
import argparse
import mimetypes
import boto3
from botocore.exceptions import ClientError

def upload_to_s3(local_directory, bucket, destination):
    try:
        client = boto3.client('s3', config=boto3.session.Config(signature_version='s3v4'))
    except ClientError as err:
        print('Failed to create boto3 client.\n' + str(err))
        return False
    try:
        uploadDirectory(client, local_directory, bucket, destination)
    except ClientError as err:
        print('Failed to upload artefact to S3.\n' + str(err))
        return False
    except IOError as err:
        print('Failed to access artefact in this directory.\n' + str(err))
        return False
    return True


def uploadDirectory(client, local_directory, bucket, destination):
    for root, dirs, files in os.walk(local_directory):
      for filename in files:
        local_path = os.path.join(root, filename)
        relative_path = os.path.relpath(local_path, local_directory)
        s3_path = relative_path

        # get content type from file for AWS metadata, with fallback
        content_type = mimetypes.guess_type(filename)[0]
        content_type = 'text/plain' if content_type is None else content_type

        try:
            client.head_object(Bucket=bucket, Key=s3_path)
            print('Path found on S3! Deleting {}...'.format(s3_path))

            try:
                client.delete_object(Bucket = bucket, Key = s3_path)
            except:
                print('Unable to delete {}...'.format(s3_path))
        except:
            print('Uploading {}...'.format(s3_path))
            client.upload_file(
                local_path,
                bucket,
                s3_path,
                ExtraArgs = { 'ContentType': content_type }
            )


def main():

    parser = argparse.ArgumentParser()
    parser.add_argument('bucket', help='Name of the existing S3 bucket')
    parser.add_argument('artefact', help='Name of the artefact to be uploaded to S3')
    parser.add_argument('bucket_key', help='Name of the S3 Bucket key')
    args = parser.parse_args()

    if not upload_to_s3(args.bucket, args.artefact, args.bucket_key):
        sys.exit(1)

if __name__ == '__main__':
    main()