boto3/boto3_util.py

import boto3
from PIL import Image
import numpy as np
from io import BytesIO
class S3Util:
    @classmethod
    def read_image_from_s3(cls, bucket, key, region_name='ap-southeast-1')->np.ndarray:
        s3 = boto3.resource('s3', region_name='ap-southeast-1')
        bucket = s3.Bucket(bucket)
        object = bucket.Object(key)
        response = object.get()
        file_stream = response['Body']
        im = Image.open(file_stream)
        return np.array(im)

    @classmethod
    def write_image_to_s3(cls, img_array, bucket, key, region_name='ap-southeast-1')->None:
        s3 = boto3.resource('s3', region_name)
        bucket = s3.Bucket(bucket)
        object = bucket.Object(key)
        file_stream = BytesIO()
        im = Image.fromarray(img_array)
        im.save(file_stream, format='jpeg')
        object.put(Body=file_stream.getvalue())