diff --git a/api/ovdashboard_api/routers/v1/_common.py b/api/ovdashboard_api/routers/v1/_common.py index 4fa8ece..e021df1 100644 --- a/api/ovdashboard_api/routers/v1/_common.py +++ b/api/ovdashboard_api/routers/v1/_common.py @@ -3,9 +3,8 @@ Dependables for defining Routers. """ import re -from dataclasses import dataclass from logging import getLogger -from typing import Iterator, Protocol +from typing import Awaitable, Callable from fastapi import Depends, HTTPException, status from webdav3.exceptions import RemoteResourceNotFound @@ -17,15 +16,6 @@ from ...core.webdav import WebDAV _logger = getLogger(__name__) -class NameLister(Protocol): - """ - Can be called to create an iterator containing some names. - """ - - async def __call__(self) -> Iterator[str]: - ... - - _RESPONSE_OK = { status.HTTP_200_OK: { "description": "Operation successful", @@ -33,47 +23,6 @@ _RESPONSE_OK = { } -@dataclass(frozen=True, slots=True) -class FileNameLister: - """ - Can be called to create an iterator containing file names. - - File names listed will be in `remote_path` and will match the RegEx `re`. - """ - - path_name: str - re: re.Pattern[str] - - @property - def responses(self) -> dict: - return { - **_RESPONSE_OK, - status.HTTP_404_NOT_FOUND: { - "description": f"{self.path_name!r} not found", - "content": None, - }, - } - - @property - async def remote_path(self) -> str: - cfg = await get_config() - - return getattr(cfg, self.path_name) - - async def __call__(self) -> Iterator[str]: - try: - file_names = await WebDAV.list_files(await self.remote_path, regex=self.re) - - return iter(file_names) - - except RemoteResourceNotFound: - _logger.error( - "WebDAV path %s lost!", - repr(await self.remote_path), - ) - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) - - async def get_remote_path( path_name: str, *, @@ -82,6 +31,18 @@ async def get_remote_path( 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], *, @@ -101,16 +62,6 @@ async def list_files( raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) -@dataclass(frozen=True, slots=True) -class CalendarNameLister: - """ - Can be called to create an iterator containing calendar names. - """ - - async def __call__(self) -> Iterator[str]: - return iter(await CalDAV.calendars) - - async def list_calendar_names() -> list[str]: """ List calendar names @@ -118,18 +69,6 @@ async def list_calendar_names() -> list[str]: return await CalDAV.calendars -@dataclass(frozen=True, slots=True) -class AggregateNameLister: - """ - Can be called to create an iterator containing aggregate calendar names. - """ - - async def __call__(self) -> Iterator[str]: - cfg = await get_config() - - return iter(cfg.calendar.aggregates.keys()) - - async def list_aggregate_names( cfg: Config = Depends(get_config), ) -> list[str]: @@ -139,71 +78,63 @@ async def list_aggregate_names( return list(cfg.calendar.aggregates.keys()) -@dataclass(frozen=True, slots=True) -class PrefixFinder: - """ - Can be called to create an iterator containing some names, all starting - with a given prefix. +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, + }, + } - All names will be taken from the list produced by the called `lister`. + +async def filter_prefix( + src: Callable[[], Awaitable[list[str]]], + prefix: str = "", +) -> list[str]: + """ + Filter names from an async source `src` for names starting with a given prefix. """ - lister: NameLister - - @property - def responses(self) -> dict: - return { - **_RESPONSE_OK, - status.HTTP_404_NOT_FOUND: { - "description": f"Failure in lister {self.lister.__class__.__name__!r}", - "content": None, - }, - } - - async def __call__(self, prefix: str) -> Iterator[str]: - return ( - file_name - for file_name in (await self.lister()) - if file_name.lower().startswith(prefix.lower()) - ) + return list( + item for item in (await src()) if item.lower().startswith(prefix.lower()) + ) -@dataclass(frozen=True, slots=True) -class PrefixUnique: +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( + src: Callable[[str], Awaitable[list[str]]], + prefix: str = "", +) -> str: """ - Can be called to determine if a given prefix is unique in the list - produced by the called `finder`. + 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. + On success, produces the unique name with that prefix. Otherwise, throws a HTTPException. """ - finder: PrefixFinder + names = await src(prefix) - @property - def responses(self) -> dict: - return { - **_RESPONSE_OK, - status.HTTP_404_NOT_FOUND: { - "description": "Prefix not found", - "content": None, - }, - status.HTTP_409_CONFLICT: { - "description": "Ambiguous prefix", - "content": None, - }, - } + match names: + case [name]: + return name - async def __call__(self, prefix: str) -> str: - names = await self.finder(prefix) - - try: - name = next(names) - - except StopIteration: + case []: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) - if any(True for _ in names): + case _: raise HTTPException(status_code=status.HTTP_409_CONFLICT) - - return name