From 44a165c53ee610886f90fff6b84b18ee81d95f51 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rn-Michael=20Miehe?= <40151420+ldericher@users.noreply.github.com> Date: Thu, 26 Oct 2023 16:31:12 +0200 Subject: [PATCH] refac: use global RP_ and LM_ objects --- api/ovdashboard_api/routers/v1/_common.py | 113 ++++++++-------------- api/ovdashboard_api/routers/v1/file.py | 30 ++---- api/ovdashboard_api/routers/v1/image.py | 31 ++---- api/ovdashboard_api/routers/v1/text.py | 33 ++----- api/ovdashboard_api/routers/v1/ticker.py | 8 +- 5 files changed, 71 insertions(+), 144 deletions(-) diff --git a/api/ovdashboard_api/routers/v1/_common.py b/api/ovdashboard_api/routers/v1/_common.py index 312525d..5bc673c 100644 --- a/api/ovdashboard_api/routers/v1/_common.py +++ b/api/ovdashboard_api/routers/v1/_common.py @@ -7,7 +7,7 @@ from dataclasses import dataclass from logging import getLogger from typing import Awaitable, Callable, Generic, ParamSpec, TypeVar -from fastapi import Depends, HTTPException, status +from fastapi import Depends, HTTPException, params, status from webdav3.exceptions import RemoteResourceNotFound from ...core.config import get_config @@ -27,12 +27,12 @@ _RESPONSE_OK = { Params = ParamSpec("Params") Return = TypeVar("Return") -type _DepCallable[**Params, Return] = Callable[Params, Awaitable[Return]] +type _Dependable[**Params, Return] = Callable[Params, Awaitable[Return]] @dataclass(slots=True, frozen=True) class Dependable(Generic[Params, Return]): - func: _DepCallable[Params, Return] + func: _Dependable[Params, Return] responses: dict @@ -52,6 +52,11 @@ class ListManager: prefix: str, names: list[str] = Depends(self.lister.func), ) -> list[str]: + if isinstance(names, params.Depends): + names = await self.lister.func() + + _logger.debug("filter %s from %s", repr(prefix), repr(names)) + return [item for item in names if item.lower().startswith(prefix.lower())] object.__setattr__( @@ -59,8 +64,14 @@ class ListManager: ) async def _getter( + prefix: str, names: list[str] = Depends(self.filter.func), ) -> str: + if isinstance(names, params.Depends): + names = await self.filter.func(prefix) + + _logger.debug("get %s from %s", repr(prefix), repr(names)) + match names: case [name]: return name @@ -93,7 +104,7 @@ class ListManager: def get_remote_path( path_name: str, -) -> _DepCallable[[], str]: +) -> _Dependable[[], str]: async def _get_remote_path() -> str: cfg = await get_config() return getattr(cfg, path_name) @@ -101,8 +112,13 @@ def get_remote_path( return _get_remote_path -def list_files( - rp: _DepCallable[[], str], +RP_FILE = get_remote_path("file_dir") +RP_IMAGE = get_remote_path("image_dir") +RP_TEXT = get_remote_path("text_dir") + + +def get_file_lister( + rp: _Dependable[[], str], *, re: re.Pattern[str], ) -> Dependable[[], list[str]]: @@ -113,6 +129,11 @@ def list_files( async def _list_files( remote_path: str = Depends(rp), ) -> list[str]: + if isinstance(remote_path, params.Depends): + remote_path = await rp() + + _logger.debug("list %s", repr(remote_path)) + try: return await WebDAV.list_files(remote_path, regex=re) @@ -132,6 +153,18 @@ def list_files( ) +LM_FILE = ListManager( + get_file_lister(rp=RP_FILE, re=re.compile(r"[^/]$", flags=re.IGNORECASE)) +) +LM_IMAGE = ListManager( + get_file_lister( + rp=RP_IMAGE, re=re.compile(r"\.(gif|jpe?g|tiff?|png|bmp)$", flags=re.IGNORECASE) + ) +) +LM_TEXT = ListManager( + get_file_lister(rp=RP_TEXT, re=re.compile(r"\.(txt|md)$", flags=re.IGNORECASE)) +) + # async def list_calendar_names() -> list[str]: # """ # List calendar names @@ -146,71 +179,3 @@ def list_files( # List aggregate calendar names # """ # return list(cfg.calendar.aggregates.keys()) - - -def filter_prefix( - src: Dependable[[], list[str]], -) -> Dependable[[str], list[str]]: - """ - Filter names from an async source `src` for names starting with a given prefix. - """ - - async def _filter_prefix( - prefix: str, - ) -> list[str]: - return list( - item - for item in (await src.func()) - 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.__class__.__name__!r}", - "content": None, - }, - }, - ) - - -def filter_prefix_unique( - src: Dependable[[str], list[str]], -) -> Dependable[[str], 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. - """ - - async def _filter_prefix_unique( - prefix: str, - ) -> str: - names = await src.func(prefix) - - match names: - case [name]: - return name - - case []: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) - - 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 5456b84..e922c4f 100644 --- a/api/ovdashboard_api/routers/v1/file.py +++ b/api/ovdashboard_api/routers/v1/file.py @@ -6,7 +6,6 @@ Router "file" provides: - getting files by name prefix """ -import re from io import BytesIO from logging import getLogger @@ -16,30 +15,19 @@ from magic import Magic from ...core.dav_common import webdav_ensure_files, webdav_ensure_path from ...core.webdav import WebDAV -from ._common import ListManager, get_remote_path, list_files +from ._common import LM_FILE, RP_FILE _logger = getLogger(__name__) _magic = Magic(mime=True) router = APIRouter(prefix="/file", tags=["file"]) -_rp = get_remote_path("file_dir") -_files = ListManager( - list_files( - rp=_rp, - re=re.compile( - r"[^/]$", - flags=re.IGNORECASE, - ), - ) -) - @router.on_event("startup") async def start_router() -> None: _logger.debug(f"{router.prefix} router starting.") - remote_path = await _rp() + remote_path = await RP_FILE() if not webdav_ensure_path(remote_path): webdav_ensure_files( remote_path, @@ -50,32 +38,32 @@ async def start_router() -> None: @router.get( "/list", - responses=_files.lister.responses, + responses=LM_FILE.lister.responses, ) async def list_all_files( - names: list[str] = Depends(_files.lister.func), + names: list[str] = Depends(LM_FILE.lister.func), ) -> list[str]: return names @router.get( "/find/{prefix}", - responses=_files.filter.responses, + responses=LM_FILE.filter.responses, ) async def find_files_by_prefix( - names: list[str] = Depends(_files.filter.func), + names: list[str] = Depends(LM_FILE.filter.func), ) -> list[str]: return names @router.get( "/get/{prefix}", - responses=_files.getter.responses, + responses=LM_FILE.getter.responses, response_class=StreamingResponse, ) async def get_file_by_prefix( - remote_path: str = Depends(_rp), - name: str = Depends(_files.getter.func), + remote_path: str = Depends(RP_FILE), + name: str = Depends(LM_FILE.getter.func), ) -> StreamingResponse: buffer = BytesIO(await WebDAV.read_bytes(f"{remote_path}/{name}")) diff --git a/api/ovdashboard_api/routers/v1/image.py b/api/ovdashboard_api/routers/v1/image.py index 811141a..f726507 100644 --- a/api/ovdashboard_api/routers/v1/image.py +++ b/api/ovdashboard_api/routers/v1/image.py @@ -6,7 +6,6 @@ Router "image" provides: - getting image files in a uniform format by name prefix """ -import re from io import BytesIO from logging import getLogger @@ -17,31 +16,19 @@ 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 filter_prefix, filter_prefix_unique, get_remote_path, list_files +from ._common import LM_IMAGE, RP_IMAGE _logger = getLogger(__name__) _PATH_NAME = "image_dir" router = APIRouter(prefix="/image", tags=["image"]) -_rp = get_remote_path(path_name=_PATH_NAME) -_ls = list_files( - rp=_rp, - re=re.compile( - r"\.(gif|jpe?g|tiff?|png|bmp)$", - flags=re.IGNORECASE, - ), -) - -_fp = filter_prefix(_ls) -_fpu = filter_prefix_unique(_fp) - @router.on_event("startup") async def start_router() -> None: _logger.debug(f"{router.prefix} router starting.") - remote_path = await _rp() + remote_path = await RP_IMAGE() if not webdav_ensure_path(remote_path): webdav_ensure_files( remote_path, @@ -53,32 +40,32 @@ async def start_router() -> None: @router.get( "/list", - responses=_ls.responses, + responses=LM_IMAGE.lister.responses, ) async def list_all_images( - names: list[str] = Depends(_ls.func), + names: list[str] = Depends(LM_IMAGE.lister.func), ) -> list[str]: return names @router.get( "/find/{prefix}", - responses=_fp.responses, + responses=LM_IMAGE.filter.responses, ) async def find_images_by_prefix( - names: list[str] = Depends(_fp.func), + names: list[str] = Depends(LM_IMAGE.filter.func), ) -> list[str]: return names @router.get( "/get/{prefix}", - responses=_fpu.responses, + responses=LM_IMAGE.getter.responses, response_class=StreamingResponse, ) async def get_image_by_prefix( - remote_path: str = Depends(_rp), - name: str = Depends(_fpu.func), + remote_path: str = Depends(RP_IMAGE), + name: str = Depends(LM_IMAGE.getter.func), ) -> StreamingResponse: cfg = await get_config() img = Image.open(BytesIO(await WebDAV.read_bytes(f"{remote_path}/{name}"))) diff --git a/api/ovdashboard_api/routers/v1/text.py b/api/ovdashboard_api/routers/v1/text.py index 12f042c..7c14a98 100644 --- a/api/ovdashboard_api/routers/v1/text.py +++ b/api/ovdashboard_api/routers/v1/text.py @@ -7,7 +7,6 @@ Router "text" provides: - getting text file HTML content by name prefix (using Markdown) """ -import re from logging import getLogger from fastapi import APIRouter, Depends @@ -15,31 +14,19 @@ from markdown import markdown from ...core.dav_common import webdav_ensure_files, webdav_ensure_path from ...core.webdav import WebDAV -from ._common import filter_prefix, filter_prefix_unique, get_remote_path, list_files +from ._common import LM_TEXT, RP_TEXT _logger = getLogger(__name__) _PATH_NAME = "text_dir" router = APIRouter(prefix="/text", tags=["text"]) -_rp = get_remote_path(path_name=_PATH_NAME) -_ls = list_files( - rp=_rp, - re=re.compile( - r"\.(txt|md)$", - flags=re.IGNORECASE, - ), -) - -_fp = filter_prefix(_ls) -_fpu = filter_prefix_unique(_fp) - @router.on_event("startup") async def start_router() -> None: _logger.debug(f"{router.prefix} router starting.") - remote_path = await _rp() + remote_path = await RP_TEXT() if not webdav_ensure_path(remote_path): webdav_ensure_files( remote_path, @@ -51,34 +38,34 @@ async def start_router() -> None: @router.get( "/list", - responses=_ls.responses, + responses=LM_TEXT.lister.responses, ) async def list_all_texts( - names: list[str] = Depends(_ls.func), + names: list[str] = Depends(LM_TEXT.lister.func), ) -> list[str]: return names @router.get( "/find/{prefix}", - responses=_fp.responses, + responses=LM_TEXT.filter.responses, ) async def find_texts_by_prefix( - names: list[str] = Depends(_fp.func), + names: list[str] = Depends(LM_TEXT.filter.func), ) -> list[str]: return names async def _get_raw_text_by_prefix( - remote_path: str = Depends(_rp), - name: str = Depends(_fpu.func), + remote_path: str = Depends(RP_TEXT), + name: str = Depends(LM_TEXT.getter.func), ) -> str: return await WebDAV.read_str(f"{remote_path}/{name}") @router.get( "/get/raw/{prefix}", - responses=_fpu.responses, + responses=LM_TEXT.getter.responses, ) async def get_raw_text_by_prefix( text: str = Depends(_get_raw_text_by_prefix), @@ -88,7 +75,7 @@ async def get_raw_text_by_prefix( @router.get( "/get/html/{prefix}", - responses=_fpu.responses, + responses=LM_TEXT.getter.responses, ) async def get_html_by_prefix( text: str = Depends(_get_raw_text_by_prefix), diff --git a/api/ovdashboard_api/routers/v1/ticker.py b/api/ovdashboard_api/routers/v1/ticker.py index 8040d0d..bbb2704 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 _fpu, _rp +from ._common import LM_TEXT, RP_TEXT _logger = getLogger(__name__) @@ -26,7 +26,7 @@ router = APIRouter(prefix="/ticker", tags=["text"]) async def start_router() -> None: _logger.debug(f"{router.prefix} router starting.") - remote_path = await _rp() + remote_path = await RP_TEXT() if not webdav_ensure_path(remote_path): webdav_ensure_files( remote_path, @@ -36,8 +36,8 @@ async def start_router() -> None: async def get_ticker_lines() -> Iterator[str]: cfg = await get_config() - file_name = await _fpu.func(cfg.ticker.file_name) - remote_path = await _rp() + file_name = await LM_TEXT.getter.func(cfg.ticker.file_name) + remote_path = await RP_TEXT() ticker = await WebDAV.read_str(f"{remote_path}/{file_name}")