""" Dependables for defining Routers. """ import re from dataclasses import dataclass from logging import getLogger from typing import Awaitable, Callable, ParamSpec, TypeVar from fastapi import HTTPException, status from webdav3.exceptions import RemoteResourceNotFound 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__) _RESPONSE_OK = { status.HTTP_200_OK: { "description": "Operation successful", }, } Params = ParamSpec("Params") Return = TypeVar("Return") @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, re: re.Pattern[str], ) -> _NDependable[list[str]]: """ List files in remote `path` matching the RegEx `re` """ async def _list_files() -> list[str]: cfg = await get_config() path = getattr(cfg, path_name) 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) return Dependable( func=_list_files, responses={ **_RESPONSE_OK, status.HTTP_404_NOT_FOUND: { "description": f"{path_name!r} not found", "content": None, }, }, ) # 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: _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( 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.__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(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, }, }, )