s3_utils.py

import hashlib
from enum import Enum

import boto3
from botocore.client import BaseClient


# S3 HANDLING ######################################################################################
def get_md5(fpath):
    with open(fpath, "rb") as f:
        file_hash = hashlib.md5()
        while chunk := f.read(8192):
            file_hash.update(chunk)
    return file_hash.hexdigest()


def upload_file_to_bucket(s3_client, file_obj, bucket, s3key):
    """Upload a file to an S3 bucket
    :param file_obj: File to upload
    :param bucket: Bucket to upload to
    :param s3key: s3key
    :param object_name: S3 object name. If not specified then file_name is used
    :return: True if file was uploaded, else False
    """
    # Upload the file
    return s3_client.upload_fileobj(
        file_obj, bucket, s3key,
        ExtraArgs={"ACL": "public-read", "ContentType": "Content-Type: audio/mpeg"}
    )


def s3_auth(aws_access_key_id, aws_secret_access_key, region_name) -> BaseClient:
    s3 = boto3.client(
        service_name='s3',
        aws_access_key_id=aws_access_key_id,
        aws_secret_access_key=aws_secret_access_key,
        region_name=region_name
    )
    return s3


def get_list_of_buckets(s3: BaseClient):
    response = s3.list_buckets()
    buckets = {}

    for buckets in response['Buckets']:
        buckets[response['Name']] = response['Name']

    BucketName = Enum('BucketName', buckets)
    return BucketName


if __name__ == '__main__':
    import os

    AWS_ACCESS_KEY_ID = os.environ['AWS_ACCESS_KEY_ID']
    AWS_SECRET_ACCESS_KEY = os.environ['AWS_SECRET_ACCESS_KEY']
    S3_BUCKET = "synthia-research"
    S3_FOLDER = "huggingface_spaces_demo"
    AWS_REGION = "eu-west-3"

    s3 = s3_auth(AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION)
    print(s3.list_buckets())

    s3key = f'{S3_FOLDER}/015.WAV'
    print(upload_file_to_bucket(s3, file_obj, S3_BUCKET, s3key))