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))