Spaces:
Sleeping
Sleeping
File size: 2,846 Bytes
ab2ae08 ed2e4a9 ab2ae08 ed2e4a9 ab2ae08 ed2e4a9 ab2ae08 ed2e4a9 ab2ae08 e76ee44 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 |
import os
import boto3
import botocore
import logging
from dotenv import load_dotenv
from demo.qccamapi.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):
res = Response()
try:
self.client.upload_fileobj(file, bucket_name, name)
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]
|