""" 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 ...core.caldav import CalDAV from ...core.config import get_config 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", }, } @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) @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) @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()) @dataclass(frozen=True, slots=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": 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()) ) @dataclass(frozen=True, slots=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