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 | |