Spaces:
Sleeping
Sleeping
import os | |
import boto3 | |
from PIL import Image | |
from io import BytesIO | |
class AwsService: | |
def session(): | |
return boto3.Session( | |
aws_access_key_id = os.environ.get('AWS_ACCESS_KEY_ID'), | |
aws_secret_access_key = os.environ.get('AWS_SECRET_ACCESS_KEY'), | |
region_name=os.environ.get('AWS_REGION') | |
) | |
def s3_client(): | |
return AwsService.session().client('s3') | |
def get_files_from_s3(bucket, prefix): | |
results = AwsService.s3_client().list_objects(Bucket=bucket, Prefix=prefix) | |
if 'Contents' in results: | |
return results['Contents'] | |
else: | |
return [] | |
def get_image_from_s3(bucket, key): | |
file_byte_string = AwsService.s3_client().get_object(Bucket=bucket, Key=key)['Body'].read() | |
return { | |
'key': key.split('/')[-1].split('.')[0], | |
'pil': Image.open(BytesIO(file_byte_string)) | |
} | |
def send_image_to_s3(file, bucket, key): | |
return AwsService.s3_client().upload_file(file, bucket, key, ExtraArgs={'ACL':'public-read', 'ContentType': 'image/png'}) |