api/ovdashboard_api/routers/v1/_common.py

@@ -3,168 +3,138 @@ Dependables for defining Routers.
 """
 
 import re
-from dataclasses import dataclass
 from logging import getLogger
-from typing import Awaitable, Callable, ParamSpec, TypeVar
+from typing import Awaitable, Callable
 
-from fastapi import HTTPException, status
+from fastapi import Depends, HTTPException, status
 from webdav3.exceptions import RemoteResourceNotFound
 
-from ...core.config import get_config
+from ...core.caldav import CalDAV
+from ...core.config import Config, get_config
 from ...core.webdav import WebDAV
 
-# from ...core.caldav import CalDAV
-# from ...core.config import Config, get_config
-
 _logger = getLogger(__name__)
 
+
 _RESPONSE_OK = {
     status.HTTP_200_OK: {
         "description": "Operation successful",
     },
 }
 
-Params = ParamSpec("Params")
-Return = TypeVar("Return")
 
-
+async def get_remote_path(
-@dataclass(slots=True, frozen=True)
-class Dependable[**Params, Return]:
-    func: Callable[Params, Awaitable[Return]]
-    responses: dict
-
-    async def __call__(self, *args: Params.args, **kwds: Params.kwargs) -> Return:
-        return await self.func(*args, **kwds)
-
-
-type _NDependable[Return] = Dependable[[], Return]
-
-
-def get_remote_path(
     path_name: str,
-) -> _NDependable[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(
     *,
-    path_name: str,
+    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(
     re: re.Pattern[str],
-) -> _NDependable[list[str]]:
+    *,
+    path: str = Depends(get_remote_path),
+) -> list[str]:
     """
     List files in remote `path` matching the RegEx `re`
     """
+    try:
+        return await WebDAV.list_files(path, regex=re)
 
-    async def _list_files() -> list[str]:
+    except RemoteResourceNotFound:
-        cfg = await get_config()
+        _logger.error(
-        path = getattr(cfg, path_name)
+            "WebDAV path %s lost!",
+            repr(path),
+        )
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
 
-        try:
-            return await WebDAV.list_files(path, regex=re)
 
-        except RemoteResourceNotFound:
+async def list_calendar_names() -> list[str]:
-            _logger.error(
+    """
-                "WebDAV path %s lost!",
+    List calendar names
-                repr(path),
+    """
-            )
+    return await CalDAV.calendars
-            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
 
-    return Dependable(
+
-        func=_list_files,
+async def list_aggregate_names(
-        responses={
+    cfg: Config = Depends(get_config),
-            **_RESPONSE_OK,
+) -> list[str]:
-            status.HTTP_404_NOT_FOUND: {
+    """
-                "description": f"{path_name!r} not found",
+    List aggregate calendar names
-                "content": None,
+    """
-            },
+    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,
         },
-    )
+    }
 
 
-# async def list_calendar_names() -> list[str]:
+async def filter_prefix(
-#     """
+    src: Callable[[], Awaitable[list[str]]],
-#     List calendar names
+    prefix: str = "",
-#     """
+) -> list[str]:
-#     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: _NDependable[list[str]],
-) -> Dependable[[str], list[str]]:
     """
     Filter names from an async source `src` for names starting with a given prefix.
     """
 
-    async def _filter_prefix(
+    return list(
-        prefix: str,
+        item for item in (await src()) if item.lower().startswith(prefix.lower())
-    ) -> 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.__class__.__name__!r}",
-                "content": None,
-            },
-        },
     )
 
 
-def filter_prefix_unique(
+def get_fpu_responses() -> dict:
-    src: Dependable[[str], list[str]],
+    return {
-) -> Dependable[[str], str]:
+        **_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(
+    src: Callable[[str], Awaitable[list[str]]],
+    prefix: 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(
+    names = await src(prefix)
-        prefix: str,
-    ) -> str:
-        names = await src(prefix)
 
-        match names:
+    match names:
-            case [name]:
+        case [name]:
-                return name
+            return name
 
-            case []:
+        case []:
-                raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
 
-            case _:
+        case _:
-                raise HTTPException(status_code=status.HTTP_409_CONFLICT)
+            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,
-            },
-        },
-    )

api/ovdashboard_api/routers/v1/file.py

@@ -9,6 +9,7 @@ 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
@@ -16,35 +17,32 @@ from magic import Magic
 
 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 FileNameLister, PrefixFinder, PrefixUnique
 
 _logger = getLogger(__name__)
 _magic = Magic(mime=True)
-_PATH_NAME = "file_dir"
 
 router = APIRouter(prefix="/file", tags=["file"])
 
-_ls = list_files(
+file_lister = FileNameLister(
-    path_name=_PATH_NAME,
+    path_name="file_dir",
     re=re.compile(
         r"[^/]$",
         flags=re.IGNORECASE,
     ),
 )
 
-_rp = get_remote_path(path_name=_PATH_NAME)
+file_finder = PrefixFinder(file_lister)
-_fp = filter_prefix(_ls)
+file_unique = PrefixUnique(file_finder)
-_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()
+    if not webdav_ensure_path(await file_lister.remote_path):
-    if not webdav_ensure_path(remote_path):
         webdav_ensure_files(
-            remote_path,
+            await file_lister.remote_path,
             "logo.svg",
             "thw.svg",
         )
@@ -53,36 +51,35 @@ async def start_router() -> None:
 @router.get(
     "/list",
     response_model=list[str],
-    responses=_ls.responses,
+    responses=file_lister.responses,
 )
-async def list_all_files(
+async def list_files(
-    names: list[str] = Depends(_ls.func),
+    names: Iterator[str] = Depends(file_lister),
 ) -> list[str]:
-    return names
+    return list(names)
 
 
 @router.get(
     "/find/{prefix}",
     response_model=list[str],
-    responses=_fp.responses,
+    responses=file_finder.responses,
 )
-async def find_files_by_prefix(
+async def find_files(
-    names: list[str] = Depends(_fp.func),
+    names: Iterator[str] = Depends(file_finder),
 ) -> list[str]:
-    return names
+    return list(names)
 
 
 @router.get(
     "/get/{prefix}",
     response_class=StreamingResponse,
-    responses=_fpu.responses,
+    responses=file_unique.responses,
 )
-async def get_file_by_prefix(
+async def get_file(
     prefix: str,
-    remote_path: str = Depends(_rp.func),
+    name: str = Depends(file_unique),
-    name: str = Depends(_fpu.func),
 ) -> StreamingResponse:
-    buffer = BytesIO(await WebDAV.read_bytes(f"{remote_path}/{name}"))
+    buffer = BytesIO(await WebDAV.read_bytes(f"{await file_lister.remote_path}/{name}"))
 
     mime = _magic.from_buffer(buffer.read(2048))
     buffer.seek(0)

api/ovdashboard_api/routers/v1/image.py

@@ -9,6 +9,7 @@ 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
@@ -17,34 +18,31 @@ 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 FileNameLister, PrefixFinder, PrefixUnique
 
 _logger = getLogger(__name__)
-_PATH_NAME = "image_dir"
 
 router = APIRouter(prefix="/image", tags=["image"])
 
-_ls = list_files(
+image_lister = FileNameLister(
-    path_name=_PATH_NAME,
+    path_name="image_dir",
     re=re.compile(
         r"\.(gif|jpe?g|tiff?|png|bmp)$",
         flags=re.IGNORECASE,
     ),
 )
 
-_rp = get_remote_path(path_name=_PATH_NAME)
+image_finder = PrefixFinder(image_lister)
-_fp = filter_prefix(_ls)
+image_unique = PrefixUnique(image_finder)
-_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()
+    if not webdav_ensure_path(await image_lister.remote_path):
-    if not webdav_ensure_path(remote_path):
         webdav_ensure_files(
-            remote_path,
+            await image_lister.remote_path,
             "img1.jpg",
             "img2.jpg",
             "img3.jpg",
@@ -54,37 +52,38 @@ async def start_router() -> None:
 @router.get(
     "/list",
     response_model=list[str],
-    responses=_ls.responses,
+    responses=image_lister.responses,
 )
-async def list_all_images(
+async def list_images(
-    names: list[str] = Depends(_ls.func),
+    names: Iterator[str] = Depends(image_lister),
 ) -> list[str]:
-    return names
+    return list(names)
 
 
 @router.get(
     "/find/{prefix}",
     response_model=list[str],
-    responses=_fp.responses,
+    responses=image_finder.responses,
 )
-async def find_images_by_prefix(
+async def find_images(
-    names: list[str] = Depends(_fp.func),
+    names: Iterator[str] = Depends(image_finder),
 ) -> list[str]:
-    return names
+    return list(names)
 
 
 @router.get(
     "/get/{prefix}",
     response_class=StreamingResponse,
-    responses=_fpu.responses,
+    responses=image_unique.responses,
 )
-async def get_image_by_prefix(
+async def get_image(
     prefix: str,
-    remote_path: str = Depends(_rp.func),
+    name: str = Depends(image_unique),
-    name: str = Depends(_fpu.func),
 ) -> StreamingResponse:
     cfg = await get_config()
-    img = Image.open(BytesIO(await WebDAV.read_bytes(f"{remote_path}/{name}")))
+    img = Image.open(
+        BytesIO(await WebDAV.read_bytes(f"{await image_lister.remote_path}/{name}"))
+    )
 
     img_buffer = BytesIO()
     img.save(img_buffer, **cfg.image.save_params)

api/ovdashboard_api/routers/v1/text.py

@@ -9,92 +9,90 @@ 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 filter_prefix, filter_prefix_unique, get_remote_path, list_files
+from ._common import FileNameLister, PrefixFinder, PrefixUnique
 
 _logger = getLogger(__name__)
-_PATH_NAME = "text_dir"
 
 router = APIRouter(prefix="/text", tags=["text"])
 
-_ls = list_files(
+text_lister = FileNameLister(
-    path_name=_PATH_NAME,
+    path_name="text_dir",
     re=re.compile(
         r"\.(txt|md)$",
         flags=re.IGNORECASE,
     ),
 )
 
-_rp = get_remote_path(path_name=_PATH_NAME)
+text_finder = PrefixFinder(text_lister)
-_fp = filter_prefix(_ls)
+text_unique = PrefixUnique(text_finder)
-_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()
+    webdav_ensure_path(await text_lister.remote_path)
-    if not webdav_ensure_path(remote_path):
+
-        webdav_ensure_files(
+    webdav_ensure_files(
-            remote_path,
+        await text_lister.remote_path,
-            "message.txt",
+        "message.txt",
-            "title.txt",
+        "title.txt",
-            "ticker.txt",
+        "ticker.txt",
-        )
+    )
 
 
 @router.get(
     "/list",
     response_model=list[str],
-    responses=_ls.responses,
+    responses=text_lister.responses,
 )
-async def list_all_texts(
+async def list_texts(
-    names: list[str] = Depends(_ls.func),
+    names: Iterator[str] = Depends(text_lister),
 ) -> list[str]:
-    return names
+    return list(names)
 
 
 @router.get(
     "/find/{prefix}",
     response_model=list[str],
-    responses=_fp.responses,
+    responses=text_finder.responses,
 )
-async def find_texts_by_prefix(
+async def find_texts(
-    names: list[str] = Depends(_fp.func),
+    names: Iterator[str] = Depends(text_finder),
 ) -> list[str]:
-    return names
+    return list(names)
 
 
-async def _get_raw_text_by_prefix(
+async def get_text_content(
-    remote_path: str = Depends(_rp.func),
+    name: str = Depends(text_unique),
-    name: str = Depends(_fpu.func),
 ) -> str:
-    return await WebDAV.read_str(f"{remote_path}/{name}")
+    return await WebDAV.read_str(f"{await text_lister.remote_path}/{name}")
-
-
-@router.get(
-    "/get/raw/{prefix}",
-    response_model=str,
-    responses=_fpu.responses,
-)
-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,
+    responses=text_unique.responses,
 )
-async def get_html_by_prefix(
+async def get_text(
-    text: str = Depends(_get_raw_text_by_prefix),
+    text: str = Depends(get_text_content),
 ) -> str:
     return markdown(text)
+
+
+@router.get(
+    "/get/raw/{prefix}",
+    response_model=str,
+    responses=text_unique.responses,
+)
+async def get_raw_text(
+    text: str = Depends(get_text_content),
+) -> str:
+    return text

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 .text import text_lister, text_unique
 
 _logger = getLogger(__name__)
 
@@ -26,20 +26,19 @@ router = APIRouter(prefix="/ticker", tags=["text"])
 async def start_router() -> None:
     _logger.debug(f"{router.prefix} router starting.")
 
-    remote_path = await _rp()
+    webdav_ensure_path(await text_lister.remote_path)
-    if not webdav_ensure_path(remote_path):
+
-        webdav_ensure_files(
+    webdav_ensure_files(
-            remote_path,
+        await text_lister.remote_path,
-            "ticker.txt",
+        "ticker.txt",
-        )
+    )
 
 
 async def get_ticker_lines() -> Iterator[str]:
     cfg = await get_config()
-    file_name = await _fpu(cfg.ticker.file_name)
+    file_name = await text_unique(cfg.ticker.file_name)
-    remote_path = await _rp()
 
-    ticker = await WebDAV.read_str(f"{remote_path}/{file_name}")
+    ticker = await WebDAV.read_str(f"{await text_lister.remote_path}/{file_name}")
 
     return (line.strip() for line in ticker.split("\n") if line.strip())