import pathlib import uuid import boto3 from fastapi import HTTPException, UploadFile, status from app.core import settings ALLOWED_CONTENT_TYPES = { "image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp", } def s3_client(): return boto3.client( "s3", endpoint_url=settings.s3_endpoint_url, aws_access_key_id=settings.s3_access_key_id, aws_secret_access_key=settings.s3_secret_access_key, region_name=settings.s3_region, ) async def upload_catalog_image( file: UploadFile, user_id: uuid.UUID, entity_type: str, ) -> dict[str, str]: if file.content_type not in ALLOWED_CONTENT_TYPES: raise HTTPException( status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, detail="Only JPEG, PNG and WEBP images are supported", ) content = await file.read() if len(content) > settings.max_upload_bytes: raise HTTPException( status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail="File is too large", ) extension = ALLOWED_CONTENT_TYPES[file.content_type] original_stem = pathlib.Path(file.filename or "image").stem[:48] object_key = f"users/{user_id}/{entity_type}/pending/{uuid.uuid4()}-{original_stem}{extension}" s3_client().put_object( Bucket=settings.s3_bucket, Key=object_key, Body=content, ContentType=file.content_type, ) public_base = settings.s3_public_base_url.rstrip("/") return { "image_s3_key": object_key, "image_s3_url": f"{public_base}/{settings.s3_bucket}/{object_key}", }