from enum import Enum from typing import Optional, Dict, Any import os from datetime import datetime from urllib.parse import urlparse import boto3 from botocore.config import Config from botocore.exceptions import ClientError from fastapi import HTTPException from pydantic_settings import BaseSettings from enums.enums import S3FolderNameEnum from exceptions.business_exception import BusinessValidationException class Settings(BaseSettings): AWS_REGION: str AWS_ACCESS_KEY: str AWS_SECRET_KEY: str AWS_BUCKET_NAME: str AWS_S3_EXPIRES: int = 60 * 60 # Default 1 hour class Config: env_file = ".env" extra = "ignore" # Allow extra fields from environment class S3Service: def __init__(self): self.settings = Settings() self.bucket_name = self.settings.AWS_BUCKET_NAME self.s3 = boto3.client( 's3', region_name=self.settings.AWS_REGION, aws_access_key_id=self.settings.AWS_ACCESS_KEY, aws_secret_access_key=self.settings.AWS_SECRET_KEY, config=Config(signature_version='s3v4') ) def get_s3_service(): return S3Service() async def upload_file( user_id: str, folder: S3FolderNameEnum, file_name: str, clinic_id: Optional[str] = None ) -> Dict[str, str]: """ Generate a pre-signed URL for uploading a file to S3. Args: user_id: The ID of the user folder: The folder enum to store the file in file_name: The name of the file clinic_id: Optional design ID for assets Returns: Dict containing the URLs and key information """ s3_service = get_s3_service() if folder == S3FolderNameEnum.ASSETS and not clinic_id: raise BusinessValidationException("Clinic id is required!") if folder != S3FolderNameEnum.PROFILE and not user_id: raise BusinessValidationException("User id is required!") timestamp = int(datetime.now().timestamp() * 1000) if folder == S3FolderNameEnum.PROFILE: key = f"common/{S3FolderNameEnum.PROFILE.value}/{user_id}/{timestamp}_{file_name}" else: key = f"common/{S3FolderNameEnum.ASSETS.value}/clinic/{clinic_id}/{timestamp}_{file_name}" try: put_url = s3_service.s3.generate_presigned_url( ClientMethod='put_object', Params={ 'Bucket': s3_service.bucket_name, 'Key': key, }, ExpiresIn=s3_service.settings.AWS_S3_EXPIRES ) get_url = s3_service.s3.generate_presigned_url( ClientMethod='get_object', Params={ 'Bucket': s3_service.bucket_name, 'Key': key, }, ExpiresIn=s3_service.settings.AWS_S3_EXPIRES ) url = urlparse(put_url) return { "api_url": put_url, "key": key, "location": f"{url.scheme}://{url.netloc}/{key}", "get_url": get_url, } except ClientError as e: print(f"Error generating pre-signed URL: {e}") raise BusinessValidationException(str(e)) def get_signed_url(key: str) -> str: """ Generate a pre-signed URL for retrieving a file from S3. Args: key: The key of the file in S3 Returns: The pre-signed URL for getting the object """ s3_service = get_s3_service() try: url = s3_service.s3.generate_presigned_url( ClientMethod='get_object', Params={ 'Bucket': s3_service.bucket_name, 'Key': key, }, ExpiresIn=3600 # 1 hour ) return url except ClientError as e: print(f"Error in get_signed_url: {e}") raise BusinessValidationException(str(e)) def get_file_key(url: str) -> str: """ Extract the file key from a URL or return the key if already provided. Args: url: The URL or key Returns: The file key """ try: if not url.startswith("http://") and not url.startswith("https://"): return url parsed_url = urlparse(url) return parsed_url.path.lstrip('/') except Exception as e: print(f"Error in get_file_key: {e}") raise BusinessValidationException(str(e))