""" Dependables for defining Routers. """ import re from dataclasses import dataclass from logging import getLogger from typing import Iterator, Protocol from fastapi import HTTPException, status from webdav3.exceptions import RemoteResourceNotFound from ..config import Config from ..dav_common import caldav_list, webdav_list _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", }, } @dataclass(frozen=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 Config.get() return str(cfg.dict()[self.path_name]) async def __call__(self) -> Iterator[str]: try: file_names = await webdav_list(await self.remote_path) return ( name for name in file_names if self.re.search(name) ) except RemoteResourceNotFound: _logger.error( "WebDAV path %s lost!", repr(await self.remote_path), ) raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) @dataclass(frozen=True) class CalendarNameLister: """ Can be called to create an iterator containing calendar names. """ async def __call__(self) -> Iterator[str]: return await caldav_list() @dataclass(frozen=True) class AggregateNameLister: """ Can be called to create an iterator containing aggregate calendar names. """ async def __call__(self) -> Iterator[str]: cfg = await Config.get() return iter(cfg.calendar.aggregates.keys()) @dataclass(frozen=True) class PrefixFinder: """ Can be called to create an iterator containing some names, all starting with a given prefix. All names will be taken from the list produced by the called `lister`. """ lister: NameLister @property def responses(self) -> dict: return { **_RESPONSE_OK, status.HTTP_404_NOT_FOUND: { "description": "Failure in lister " + repr(self.lister.__class__.__name__), "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()) ) @dataclass(frozen=True) class PrefixUnique: """ Can be called to determine if a given prefix is unique in the list produced by the called `finder`. On success, produces the unique name with that prefix. Otherwise, throws a HTTPException. """ finder: PrefixFinder @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, }, } async def __call__(self, prefix: str) -> str: names = await self.finder(prefix) try: name = next(names) except StopIteration: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) if any(True for _ in names): raise HTTPException(status_code=status.HTTP_409_CONFLICT) return name