""" Dependables for defining Routers. """ import re from dataclasses import dataclass from typing import Iterator, Protocol from fastapi import HTTPException, status from webdav3.exceptions import RemoteResourceNotFound from ..dav_common import caldav_principal, webdav_list @dataclass(frozen=True) 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`. """ remote_path: str re: re.Pattern[str] @property def responses(self) -> dict: return { **_RESPONSE_OK, status.HTTP_404_NOT_FOUND: { "description": f"{self.remote_path!r} not found", "content": None, }, } async def __call__(self) -> Iterator[str]: try: file_names = await webdav_list(self.remote_path) return ( name for name in file_names if self.re.search(name) ) except RemoteResourceNotFound: 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 ( cal.name for cal in caldav_principal().calendars() ) @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