""" Router "file" provides: - listing files - finding files by name prefix - getting files by name prefix """ import re from io import BytesIO from logging import getLogger from typing import Iterator from fastapi import APIRouter, Depends from fastapi.responses import StreamingResponse from magic import Magic from ...dav_common import webdav_ensure_files, webdav_ensure_path from ...dav_file import DavFile from ._common import FileNameLister, PrefixFinder, PrefixUnique _logger = getLogger(__name__) _magic = Magic(mime=True) router = APIRouter(prefix="/file", tags=["file"]) file_lister = FileNameLister( path_name="file_dir", re=re.compile( r"[^/]$", flags=re.IGNORECASE, ), ) file_finder = PrefixFinder(file_lister) file_unique = PrefixUnique(file_finder) @router.on_event("startup") async def start_router() -> None: _logger.debug(f"{router.prefix} router starting.") if not webdav_ensure_path(await file_lister.remote_path): webdav_ensure_files( await file_lister.remote_path, "logo.svg", ) @router.get( "/list", response_model=list[str], responses=file_lister.responses, ) async def list_files( names: Iterator[str] = Depends(file_lister), ) -> list[str]: return list(names) @router.get( "/find/{prefix}", response_model=list[str], responses=file_finder.responses, ) async def find_files( names: Iterator[str] = Depends(file_finder), ) -> list[str]: return list(names) @router.get( "/get/{prefix}", response_class=StreamingResponse, responses=file_unique.responses, ) async def get_file( prefix: str, name: str = Depends(file_unique), ) -> StreamingResponse: dav_file = DavFile(f"{await file_lister.remote_path}/{name}") buffer = BytesIO(await dav_file.as_bytes) mime = _magic.from_buffer(buffer.read(2048)) buffer.seek(0) return StreamingResponse( content=buffer, media_type=mime, headers={ "Content-Disposition": f"filename={prefix}" }, )