apis/s3.py

import os
import boto3
import botocore
import logging
import json
from dotenv import load_dotenv

from models.response import Response
load_dotenv()

class S3:

    def __init__(self, custom_logging=None) -> None:
        self.logging = custom_logging if custom_logging is not None else logging

        self.aws_access_key_id = os.getenv('AWS_ACCESS_KEY_ID', '')
        self.aws_secret_access_key = os.getenv('AWS_SECRET_ACCESS_KEY', '')
        self.aws_default_region = os.getenv('AWS_DEFAULT_REGION', '')

        self.client = boto3.client(
            service_name='s3',
            region_name=self.aws_default_region,
            aws_access_key_id=self.aws_access_key_id,
            aws_secret_access_key=self.aws_secret_access_key
        )

    # Function to list images in a specific S3 bucket
    def list_images_in_bucket(self, bucket_name):
        response = self.client.list_objects_v2(Bucket=bucket_name)
        return [obj['Key'] for obj in response.get('Contents', []) if obj['Key'].lower().endswith(('.png', '.jpg', '.jpeg'))]
    
    # Function to get image URLs
    def get_image_urls(self, bucket_name, image_keys):
        return [self.client.generate_presigned_url('get_object', Params={'Bucket': bucket_name, 'Key': key}) for key in image_keys]

    # Function to upload file to S3
    def upload_file(self, bucket_name, file, name, metadata:dict):
        res = Response()
        try:
            # self.client.upload_fileobj(
            #     Fileobj=file, 
            #     Bucket=bucket_name, 
            #     Key=name, 
            #     ExtraArgs={
            #         'Metadata': {
            #             'x-request-id': req_id,
            #         }
            #     })
            
            metadata = {k: str(v) for k, v in metadata.items()}

            self.client.put_object(
                Body=file, 
                Bucket=bucket_name, 
                Key=name, 
                Metadata=metadata
            )
            res.status = 200
        except botocore.exceptions.ClientError as error:
            res.status = 429
            if error.response['Error']['Code'] == 'LimitExceededException':
                res.error = ValueError('API call limit exceeded; backing off and retrying...')
            else:
                res.error = error
        except botocore.exceptions.ParamValidationError as error:
            res.status = 400
            res.error = ValueError('The parameters you provided are incorrect: {}'.format(error))
        return res

    # Function to delete a file from S3
    def delete_image(self, bucket_name, file_key):
        if isinstance(file_key, list):
            for k in file_key:
                print(f'Removing "{k}"...')
                self.client.delete_object(Bucket=bucket_name, Key=k)
        else:
            self.client.delete_object(Bucket=bucket_name, Key=file_key)

    # Function to list images in a specific S3 bucket
    def list_files_in_bucket(self, bucket_name):
        response = self.client.list_objects_v2(Bucket=bucket_name)
        return [obj['Key'] for obj in response.get('Contents', [])]

    # Function to get image URLs
    def get_files_urls(self, bucket_name, file_keys):
        return [self.client.generate_presigned_url('get_object', Params={'Bucket': bucket_name, 'Key': key}) for key in file_keys]