Spaces:
Runtime error
Runtime error
from PIL.ImageFile import ImageFile | |
import boto3 | |
import tempfile | |
import os | |
import logging | |
from dotenv import load_dotenv | |
load_dotenv(verbose=True) | |
AWS_ACCESS_KEY = os.getenv("AWS_ACCESS_KEY_ID") | |
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY") | |
AWS_BUCKET_NAME = os.getenv("AWS_S3_BUCKET_NAME") | |
AWS_REGION = os.getenv("AWS_REGION") | |
class Bucket: | |
def __init__(self): | |
self.bucket_name = AWS_BUCKET_NAME | |
self.s3 = boto3.client( | |
"s3", | |
region_name=AWS_REGION, | |
aws_access_key_id=AWS_ACCESS_KEY, | |
aws_secret_access_key=AWS_SECRET_ACCESS_KEY, | |
) | |
def upload_file(self, image: ImageFile, uuid: str): | |
key = f"{uuid}.png" | |
try: | |
logging.info( | |
f"Uploading image to S3 with key: s3://{self.bucket_name}/{key}" | |
) | |
with tempfile.TemporaryFile() as fp: | |
image.save(fp, "PNG") | |
fp.seek(0) | |
self.s3.upload_fileobj(fp, self.bucket_name, key) | |
except Exception as e: | |
logging.error(e) | |
raise e | |
def get_presigned_url(self, uuid: str): | |
key = f"{uuid}.png" | |
try: | |
url = self.s3.generate_presigned_url( | |
"get_object", | |
Params={"Bucket": self.bucket_name, "Key": key}, | |
ExpiresIn=3600, | |
) | |
return url | |
except Exception as e: | |
logging.error(e) | |
raise e | |