import re from io import BytesIO from typing import Iterator from fastapi import APIRouter, Depends, HTTPException, status from fastapi.responses import StreamingResponse from PIL import Image from ..dav_file import DavFile from ._common import FileNameLister, FilePrefixFinder router = APIRouter(prefix="/image", tags=["image"]) _lister = FileNameLister( remote_path="img", re=re.compile( r"\.(gif|jpe?g|tiff?|png|bmp)$", flags=re.IGNORECASE, ), ) _finder = FilePrefixFinder(_lister) @router.get("/list", response_model=list[str]) async def list_images( image_file_names: Iterator[str] = Depends(_lister), ) -> list[str]: return list(image_file_names) @router.get("/find/{prefix}", response_model=list[str]) async def find_images( file_names: Iterator[str] = Depends(_finder), ) -> list[str]: return list(file_names) @router.get( "/get/{prefix}", response_class=StreamingResponse, responses={ status.HTTP_200_OK: { "description": "Operation successful", }, status.HTTP_404_NOT_FOUND: { "description": "image file not found", "content": None, }, status.HTTP_409_CONFLICT: { "description": "ambiguous image file name", "content": None, }, }, ) async def get_image( prefix: str, file_names: Iterator[str] = Depends(_finder), ) -> StreamingResponse: file_names = list(file_names) if not (file_names := list(file_names)): raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) elif len(file_names) > 1: raise HTTPException(status_code=status.HTTP_409_CONFLICT) img_file = DavFile(f"img/{file_names[0]}") img = Image.open(BytesIO(await img_file.bytes)).convert("RGB") img_buffer = BytesIO() img.save(img_buffer, format='JPEG', quality=85) img_buffer.seek(0) return StreamingResponse( content=img_buffer, media_type="image/jpeg", headers={ "Content-Disposition": f"filename={prefix}.jpg" }, )