Initial Commit

.gitignore

@@ -0,0 +1,4 @@
+.venv
+venv
+.idea
+__*

Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.12.9-alpine
+
+ENV APP_HOME=/home/app
+
+WORKDIR $APP_HOME
+
+COPY requirements.txt .
+
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 8000
+
+CMD ["uvicorn", "main:app", "--port", "8000", "--host", "0.0.0.0"]

docker-compose.yml

@@ -0,0 +1,14 @@
+services:
+  image_processor:
+    container_name: image_processor
+    ports:
+      - "8000:8000"
+    build:
+      context: .
+      dockerfile: Dockerfile
+    restart: unless-stopped
+
+
+
+
+

exceptions.py

@@ -0,0 +1,9 @@
+from fastapi import HTTPException, status
+
+
+class InternalServerErrorException(HTTPException):
+	status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+	detail = "Internal Server Error"
+
+	def __init__(self):
+		super().__init__(status_code=self.status_code, detail=self.detail)

image_service.py

@@ -0,0 +1,44 @@
+import io
+from urllib.parse import urljoin
+
+import aioboto3
+from PIL import Image
+
+import pillow_avif
+from pydantic import HttpUrl
+
+from schemas import SImageWithFormat
+
+SUPPORTED_FORMATS = {"avif", "png", "webp"}
+
+def process_image(img_bytes: bytes, formats: set[str]) -> list[SImageWithFormat]:
+    images = []
+    image = Image.open(io.BytesIO(img_bytes))
+    for image_format in formats:
+        buffer = io.BytesIO()
+        image.save(buffer, format=image_format)
+        images.append(SImageWithFormat.model_validate({"image": buffer.getvalue(), "format": image_format}))
+
+    return images
+
+async def upload_images(imgs: list[SImageWithFormat], name: str, upload_url: str, bucket_name: str) -> list[HttpUrl]:
+    urls = []
+    s3_session = aioboto3.Session()
+    async with s3_session.client(
+            "s3",
+            endpoint_url=upload_url,
+    ) as s3_client:
+        for img in imgs:
+            object_key = f"{name}.{img.format}"
+
+            # await s3_client.put_object(
+            #     Bucket=bucket_name,
+            #     Key=object_key,
+            #     Body=img.image,
+            #     ContentType=f"image/{img.format}"
+            # )
+
+            urls.append(HttpUrl(urljoin(upload_url, f"{bucket_name}/{object_key}")))
+
+    return urls
+

main.py

@@ -0,0 +1,52 @@
+from base64 import b64decode
+
+from fastapi import FastAPI, status
+
+from exceptions import InternalServerErrorException
+from image_service import SUPPORTED_FORMATS, process_image, upload_images
+from schemas import SUploadedImages, SFormats, SAvailableFormats, SInternalServerError
+
+app = FastAPI()
+
+
+@app.post(
+    "/process_images",
+    status_code=status.HTTP_200_OK,
+    response_model=SUploadedImages,
+    responses={
+        status.HTTP_500_INTERNAL_SERVER_ERROR: {
+			"model": SInternalServerError,
+			"description": "Internal Server Error",
+		},
+    },
+)
+async def process_images(upload_data: SFormats):
+    try:
+        all_urls = []
+        for image in upload_data.images:
+            imgs = process_image(b64decode(image.image), upload_data.formats)
+            urls = await upload_images(imgs, image.name, str(upload_data.upload_url), upload_data.bucket_name)
+            all_urls.append(urls)
+
+        return {"images_urls": all_urls}
+    except Exception as e:
+        print(e)
+        raise InternalServerErrorException
+
+
+
+@app.get(
+    "/available_formats",
+    status_code=status.HTTP_200_OK,
+    response_model=SAvailableFormats,
+    responses={
+        status.HTTP_500_INTERNAL_SERVER_ERROR: {
+			"model": SInternalServerError,
+			"description": "Internal Server Error",
+		},
+    },
+)
+async def get_available_formats():
+    return {"available_formats": SUPPORTED_FORMATS}
+
+

requirements.txt

@@ -0,0 +1,61 @@
+aioboto3==14.1.0
+aiobotocore==2.21.1
+aiofiles==24.1.0
+aiohappyeyeballs==2.5.0
+aiohttp==3.11.13
+aioitertools==0.12.0
+aiosignal==1.3.2
+annotated-types==0.7.0
+anyio==4.8.0
+attrs==25.1.0
+boto3==1.37.1
+botocore==1.37.1
+certifi==2025.1.31
+click==8.1.8
+dnspython==2.7.0
+email_validator==2.2.0
+fastapi==0.115.11
+fastapi-cli==0.0.7
+frozenlist==1.5.0
+h11==0.14.0
+httpcore==1.0.7
+httptools==0.6.4
+httpx==0.28.1
+idna==3.10
+itsdangerous==2.2.0
+Jinja2==3.1.6
+jmespath==1.0.1
+markdown-it-py==3.0.0
+MarkupSafe==3.0.2
+mdurl==0.1.2
+multidict==6.1.0
+orjson==3.10.15
+pillow==11.1.0
+pillow-avif-plugin==1.4.6
+propcache==0.3.0
+pydantic==2.10.6
+pydantic-extra-types==2.10.2
+pydantic-settings==2.8.1
+pydantic_core==2.27.2
+Pygments==2.19.1
+python-dateutil==2.9.0.post0
+python-dotenv==1.0.1
+python-multipart==0.0.20
+PyYAML==6.0.2
+rich==13.9.4
+rich-toolkit==0.13.2
+s3transfer==0.11.3
+shellingham==1.5.4
+six==1.17.0
+sniffio==1.3.1
+starlette==0.46.0
+typer==0.15.2
+typing_extensions==4.12.2
+ujson==5.10.0
+urllib3==2.3.0
+uvicorn==0.34.0
+uvloop==0.21.0
+watchfiles==1.0.4
+websockets==15.0.1
+wrapt==1.17.2
+yarl==1.18.3

schemas.py

@@ -0,0 +1,27 @@
+from typing import Literal
+
+from pydantic import BaseModel, HttpUrl
+
+
+class UploadImage(BaseModel):
+    image: bytes
+    name: str
+
+class SFormats(BaseModel):
+    formats: set[Literal["avif", "png", "webp"]]
+    images: list[UploadImage]
+    upload_url: HttpUrl
+    bucket_name: str
+
+class SUploadedImages(BaseModel):
+    images_urls: list[list[HttpUrl]]
+
+class SAvailableFormats(BaseModel):
+    available_formats: set[str]
+
+class SImageWithFormat(BaseModel):
+    image: bytes
+    format: str
+
+class SInternalServerError(BaseModel):
+    detail: str = "Internal Server Error"