from __future__ import annotations from io import BytesIO from pathlib import Path from fastapi import HTTPException, UploadFile, status from PIL import Image, UnidentifiedImageError from pydantic import BaseModel, Field, model_validator, ValidationError from .config import Settings class ImageUploadMetadata(BaseModel): filename: str content_type: str size_bytes: int = Field(gt=0) @model_validator(mode="after") def validate_metadata(self) -> "ImageUploadMetadata": allowed_extensions = {".jpg", ".jpeg", ".png", ".webp", ".bmp"} allowed_content_types = {"image/jpeg", "image/png", "image/webp", "image/bmp"} suffix = Path(self.filename).suffix.lower() if suffix not in allowed_extensions: raise ValueError("Unsupported file type") if self.content_type not in allowed_content_types: raise ValueError("Unsupported file type") return self async def read_and_validate_image(file: UploadFile, settings: Settings) -> bytes: if not file.filename: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="File name is required") data = await file.read(settings.max_upload_bytes + 1) try: metadata = ImageUploadMetadata( filename=file.filename, content_type=file.content_type or "", size_bytes=len(data), ) except ValidationError: raise HTTPException(status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, detail="Unsupported file type") if metadata.size_bytes > settings.max_upload_bytes: raise HTTPException(status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail="File exceeds 5 MB limit") try: with Image.open(BytesIO(data)) as image: image.verify() with Image.open(BytesIO(data)) as image: image.convert("RGB") except (UnidentifiedImageError, OSError, ValueError) as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Unable to decode image") from exc return data