diff --git a/api/ovdashboard_api/routers/v1/_common.py b/api/ovdashboard_api/routers/v1/_common.py index e021df1..4d3ec6c 100644 --- a/api/ovdashboard_api/routers/v1/_common.py +++ b/api/ovdashboard_api/routers/v1/_common.py @@ -3,16 +3,19 @@ Dependables for defining Routers. """ import re +from dataclasses import dataclass from logging import getLogger -from typing import Awaitable, Callable +from typing import Awaitable, Callable, ParamSpec, TypeVar -from fastapi import Depends, HTTPException, status +from fastapi import HTTPException, status from webdav3.exceptions import RemoteResourceNotFound -from ...core.caldav import CalDAV -from ...core.config import Config, get_config +from ...core.config import get_config from ...core.webdav import WebDAV +# from ...core.caldav import CalDAV +# from ...core.config import Config, get_config + _logger = getLogger(__name__) @@ -22,119 +25,141 @@ _RESPONSE_OK = { }, } +Params = ParamSpec("Params") +Return = TypeVar("Return") -async def get_remote_path( + +@dataclass(slots=True, frozen=True) +class Dependable[**Params, Return]: + func: Callable[Params, Return] + responses: dict + + +def get_remote_path( path_name: str, +) -> Dependable[[], Awaitable[str]]: + async def _get_remote_path() -> str: + cfg = await get_config() + return getattr(cfg, path_name) + + return Dependable( + func=_get_remote_path, + responses={**_RESPONSE_OK}, + ) + + +def list_files( *, - cfg: Config = Depends(get_config), -) -> str: - return getattr(cfg, path_name) - - -def get_lf_responses( - path: str = Depends(get_remote_path), -) -> dict: - return { - **_RESPONSE_OK, - status.HTTP_404_NOT_FOUND: { - "description": f"{path!r} not found", - "content": None, - }, - } - - -async def list_files( + path_name: str, re: re.Pattern[str], - *, - path: str = Depends(get_remote_path), -) -> list[str]: +) -> Dependable[[], Awaitable[list[str]]]: """ List files in remote `path` matching the RegEx `re` """ - try: - return await WebDAV.list_files(path, regex=re) - except RemoteResourceNotFound: - _logger.error( - "WebDAV path %s lost!", - repr(path), - ) - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) + async def _list_files() -> list[str]: + cfg = await get_config() + path = getattr(cfg, path_name) + try: + return await WebDAV.list_files(path, regex=re) -async def list_calendar_names() -> list[str]: - """ - List calendar names - """ - return await CalDAV.calendars + except RemoteResourceNotFound: + _logger.error( + "WebDAV path %s lost!", + repr(path), + ) + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) - -async def list_aggregate_names( - cfg: Config = Depends(get_config), -) -> list[str]: - """ - List aggregate calendar names - """ - return list(cfg.calendar.aggregates.keys()) - - -def get_fp_responses( - src: Callable[[], Awaitable[list[str]]], -) -> dict: - return { - **_RESPONSE_OK, - status.HTTP_404_NOT_FOUND: { - "description": f"Failure in lister {src.__name__!r}", - "content": None, + return Dependable( + func=_list_files, + responses={ + **_RESPONSE_OK, + status.HTTP_404_NOT_FOUND: { + "description": f"{path_name!r} not found", + "content": None, + }, }, - } + ) -async def filter_prefix( +# async def list_calendar_names() -> list[str]: +# """ +# List calendar names +# """ +# return await CalDAV.calendars + + +# async def list_aggregate_names( +# cfg: Config = Depends(get_config), +# ) -> list[str]: +# """ +# List aggregate calendar names +# """ +# return list(cfg.calendar.aggregates.keys()) + + +def filter_prefix( src: Callable[[], Awaitable[list[str]]], - prefix: str = "", -) -> list[str]: +) -> Dependable[[str], Awaitable[list[str]]]: """ Filter names from an async source `src` for names starting with a given prefix. """ - return list( - item for item in (await src()) if item.lower().startswith(prefix.lower()) + async def _filter_prefix( + prefix: str, + ) -> list[str]: + return list( + item for item in (await src()) if item.lower().startswith(prefix.lower()) + ) + + return Dependable( + func=_filter_prefix, + responses={ + **_RESPONSE_OK, + status.HTTP_404_NOT_FOUND: { + "description": f"Failure in lister {src.__name__!r}", + "content": None, + }, + }, ) -def get_fpu_responses() -> dict: - return { - **_RESPONSE_OK, - status.HTTP_404_NOT_FOUND: { - "description": "Prefix not found", - "content": None, - }, - status.HTTP_409_CONFLICT: { - "description": "Ambiguous prefix", - "content": None, - }, - } - - -async def filter_prefix_unique( +def filter_prefix_unique( src: Callable[[str], Awaitable[list[str]]], - prefix: str = "", -) -> str: +) -> Dependable[[str], Awaitable[str]]: """ Determines if a given prefix is unique in the list produced by the async source `src`. On success, produces the unique name with that prefix. Otherwise, throws a HTTPException. """ - names = await src(prefix) + async def _filter_prefix_unique( + prefix: str, + ) -> str: + names = await src(prefix) - match names: - case [name]: - return name + match names: + case [name]: + return name - case []: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) + case []: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) - case _: - raise HTTPException(status_code=status.HTTP_409_CONFLICT) + case _: + raise HTTPException(status_code=status.HTTP_409_CONFLICT) + + return Dependable( + func=_filter_prefix_unique, + responses={ + **_RESPONSE_OK, + status.HTTP_404_NOT_FOUND: { + "description": "Prefix not found", + "content": None, + }, + status.HTTP_409_CONFLICT: { + "description": "Ambiguous prefix", + "content": None, + }, + }, + ) diff --git a/api/ovdashboard_api/routers/v1/file.py b/api/ovdashboard_api/routers/v1/file.py index 4a79523..1fc3932 100644 --- a/api/ovdashboard_api/routers/v1/file.py +++ b/api/ovdashboard_api/routers/v1/file.py @@ -9,7 +9,6 @@ Router "file" provides: import re from io import BytesIO from logging import getLogger -from typing import Iterator from fastapi import APIRouter, Depends from fastapi.responses import StreamingResponse @@ -17,32 +16,35 @@ from magic import Magic from ...core.dav_common import webdav_ensure_files, webdav_ensure_path from ...core.webdav import WebDAV -from ._common import FileNameLister, PrefixFinder, PrefixUnique +from ._common import filter_prefix, filter_prefix_unique, get_remote_path, list_files _logger = getLogger(__name__) _magic = Magic(mime=True) +_PATH_NAME = "file_dir" router = APIRouter(prefix="/file", tags=["file"]) -file_lister = FileNameLister( - path_name="file_dir", +_ls = list_files( + path_name=_PATH_NAME, re=re.compile( r"[^/]$", flags=re.IGNORECASE, ), ) -file_finder = PrefixFinder(file_lister) -file_unique = PrefixUnique(file_finder) +_rp = get_remote_path(path_name=_PATH_NAME) +_fp = filter_prefix(_ls.func) +_fpu = filter_prefix_unique(_fp.func) @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): + remote_path = await _rp.func() + if not webdav_ensure_path(remote_path): webdav_ensure_files( - await file_lister.remote_path, + remote_path, "logo.svg", "thw.svg", ) @@ -51,35 +53,36 @@ async def start_router() -> None: @router.get( "/list", response_model=list[str], - responses=file_lister.responses, + responses=_ls.responses, ) -async def list_files( - names: Iterator[str] = Depends(file_lister), +async def list_all_files( + names: list[str] = Depends(_ls.func), ) -> list[str]: - return list(names) + return names @router.get( "/find/{prefix}", response_model=list[str], - responses=file_finder.responses, + responses=_fp.responses, ) -async def find_files( - names: Iterator[str] = Depends(file_finder), +async def find_files_by_prefix( + names: list[str] = Depends(_fp.func), ) -> list[str]: - return list(names) + return names @router.get( "/get/{prefix}", response_class=StreamingResponse, - responses=file_unique.responses, + responses=_fpu.responses, ) -async def get_file( +async def get_file_by_prefix( prefix: str, - name: str = Depends(file_unique), + remote_path: str = Depends(_rp.func), + name: str = Depends(_fpu.func), ) -> StreamingResponse: - buffer = BytesIO(await WebDAV.read_bytes(f"{await file_lister.remote_path}/{name}")) + buffer = BytesIO(await WebDAV.read_bytes(f"{remote_path}/{name}")) mime = _magic.from_buffer(buffer.read(2048)) buffer.seek(0) diff --git a/api/ovdashboard_api/routers/v1/image.py b/api/ovdashboard_api/routers/v1/image.py index cea6a72..145b400 100644 --- a/api/ovdashboard_api/routers/v1/image.py +++ b/api/ovdashboard_api/routers/v1/image.py @@ -9,7 +9,6 @@ Router "image" provides: import re from io import BytesIO from logging import getLogger -from typing import Iterator from fastapi import APIRouter, Depends from fastapi.responses import StreamingResponse @@ -18,31 +17,34 @@ from PIL import Image from ...core.config import Config, ImageUIConfig, get_config from ...core.dav_common import webdav_ensure_files, webdav_ensure_path from ...core.webdav import WebDAV -from ._common import FileNameLister, PrefixFinder, PrefixUnique +from ._common import filter_prefix, filter_prefix_unique, get_remote_path, list_files _logger = getLogger(__name__) +_PATH_NAME = "image_dir" router = APIRouter(prefix="/image", tags=["image"]) -image_lister = FileNameLister( - path_name="image_dir", +_ls = list_files( + path_name=_PATH_NAME, re=re.compile( r"\.(gif|jpe?g|tiff?|png|bmp)$", flags=re.IGNORECASE, ), ) -image_finder = PrefixFinder(image_lister) -image_unique = PrefixUnique(image_finder) +_rp = get_remote_path(path_name=_PATH_NAME) +_fp = filter_prefix(_ls.func) +_fpu = filter_prefix_unique(_fp.func) @router.on_event("startup") async def start_router() -> None: _logger.debug(f"{router.prefix} router starting.") - if not webdav_ensure_path(await image_lister.remote_path): + remote_path = await _rp.func() + if not webdav_ensure_path(remote_path): webdav_ensure_files( - await image_lister.remote_path, + remote_path, "img1.jpg", "img2.jpg", "img3.jpg", @@ -52,38 +54,37 @@ async def start_router() -> None: @router.get( "/list", response_model=list[str], - responses=image_lister.responses, + responses=_ls.responses, ) -async def list_images( - names: Iterator[str] = Depends(image_lister), +async def list_all_images( + names: list[str] = Depends(_ls.func), ) -> list[str]: - return list(names) + return names @router.get( "/find/{prefix}", response_model=list[str], - responses=image_finder.responses, + responses=_fp.responses, ) -async def find_images( - names: Iterator[str] = Depends(image_finder), +async def find_images_by_prefix( + names: list[str] = Depends(_fp.func), ) -> list[str]: - return list(names) + return names @router.get( "/get/{prefix}", response_class=StreamingResponse, - responses=image_unique.responses, + responses=_fpu.responses, ) -async def get_image( +async def get_image_by_prefix( prefix: str, - name: str = Depends(image_unique), + remote_path: str = Depends(_rp.func), + name: str = Depends(_fpu.func), ) -> StreamingResponse: cfg = await get_config() - img = Image.open( - BytesIO(await WebDAV.read_bytes(f"{await image_lister.remote_path}/{name}")) - ) + img = Image.open(BytesIO(await WebDAV.read_bytes(f"{remote_path}/{name}"))) img_buffer = BytesIO() img.save(img_buffer, **cfg.image.save_params) diff --git a/api/ovdashboard_api/routers/v1/text.py b/api/ovdashboard_api/routers/v1/text.py index b4a0255..9a1c239 100644 --- a/api/ovdashboard_api/routers/v1/text.py +++ b/api/ovdashboard_api/routers/v1/text.py @@ -9,90 +9,92 @@ Router "text" provides: import re from logging import getLogger -from typing import Iterator from fastapi import APIRouter, Depends from markdown import markdown from ...core.dav_common import webdav_ensure_files, webdav_ensure_path from ...core.webdav import WebDAV -from ._common import FileNameLister, PrefixFinder, PrefixUnique +from ._common import filter_prefix, filter_prefix_unique, get_remote_path, list_files _logger = getLogger(__name__) +_PATH_NAME = "text_dir" router = APIRouter(prefix="/text", tags=["text"]) -text_lister = FileNameLister( - path_name="text_dir", +_ls = list_files( + path_name=_PATH_NAME, re=re.compile( r"\.(txt|md)$", flags=re.IGNORECASE, ), ) -text_finder = PrefixFinder(text_lister) -text_unique = PrefixUnique(text_finder) +_rp = get_remote_path(path_name=_PATH_NAME) +_fp = filter_prefix(_ls.func) +_fpu = filter_prefix_unique(_fp.func) @router.on_event("startup") async def start_router() -> None: _logger.debug(f"{router.prefix} router starting.") - webdav_ensure_path(await text_lister.remote_path) - - webdav_ensure_files( - await text_lister.remote_path, - "message.txt", - "title.txt", - "ticker.txt", - ) + remote_path = await _rp.func() + if not webdav_ensure_path(remote_path): + webdav_ensure_files( + remote_path, + "message.txt", + "title.txt", + "ticker.txt", + ) @router.get( "/list", response_model=list[str], - responses=text_lister.responses, + responses=_ls.responses, ) -async def list_texts( - names: Iterator[str] = Depends(text_lister), +async def list_all_texts( + names: list[str] = Depends(_ls.func), ) -> list[str]: - return list(names) + return names @router.get( "/find/{prefix}", response_model=list[str], - responses=text_finder.responses, + responses=_fp.responses, ) -async def find_texts( - names: Iterator[str] = Depends(text_finder), +async def find_texts_by_prefix( + names: list[str] = Depends(_fp.func), ) -> list[str]: - return list(names) + return names -async def get_text_content( - name: str = Depends(text_unique), +async def _get_raw_text_by_prefix( + remote_path: str = Depends(_rp.func), + name: str = Depends(_fpu.func), ) -> str: - return await WebDAV.read_str(f"{await text_lister.remote_path}/{name}") - - -@router.get( - "/get/html/{prefix}", - response_model=str, - responses=text_unique.responses, -) -async def get_text( - text: str = Depends(get_text_content), -) -> str: - return markdown(text) + return await WebDAV.read_str(f"{remote_path}/{name}") @router.get( "/get/raw/{prefix}", response_model=str, - responses=text_unique.responses, + responses=_fpu.responses, ) -async def get_raw_text( - text: str = Depends(get_text_content), +async def get_raw_text_by_prefix( + text: str = Depends(_get_raw_text_by_prefix), ) -> str: return text + + +@router.get( + "/get/html/{prefix}", + response_model=str, + responses=_fpu.responses, +) +async def get_html_by_prefix( + text: str = Depends(_get_raw_text_by_prefix), +) -> str: + return markdown(text) diff --git a/api/ovdashboard_api/routers/v1/ticker.py b/api/ovdashboard_api/routers/v1/ticker.py index bfd9db4..bb59b62 100644 --- a/api/ovdashboard_api/routers/v1/ticker.py +++ b/api/ovdashboard_api/routers/v1/ticker.py @@ -15,7 +15,7 @@ from markdown import markdown from ...core.config import Config, TickerUIConfig, get_config from ...core.dav_common import webdav_ensure_files, webdav_ensure_path from ...core.webdav import WebDAV -from .text import text_lister, text_unique +from .text import _fpu, _rp _logger = getLogger(__name__) @@ -26,19 +26,20 @@ router = APIRouter(prefix="/ticker", tags=["text"]) async def start_router() -> None: _logger.debug(f"{router.prefix} router starting.") - webdav_ensure_path(await text_lister.remote_path) - - webdav_ensure_files( - await text_lister.remote_path, - "ticker.txt", - ) + remote_path = await _rp.func() + if not webdav_ensure_path(remote_path): + webdav_ensure_files( + remote_path, + "ticker.txt", + ) async def get_ticker_lines() -> Iterator[str]: cfg = await get_config() - file_name = await text_unique(cfg.ticker.file_name) + file_name = await _fpu.func(cfg.ticker.file_name) + remote_path = await _rp.func() - ticker = await WebDAV.read_str(f"{await text_lister.remote_path}/{file_name}") + ticker = await WebDAV.read_str(f"{remote_path}/{file_name}") return (line.strip() for line in ticker.split("\n") if line.strip())