ovdashboard/api/ovdashboard_api/routers/_common.py

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):
    def __call__(self) -> Iterator[str]:
        ...


@dataclass(frozen=True)
class FileNameLister:
    remote_path: str
    re: re.Pattern[str]

    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:
            return iter(())


@dataclass(frozen=True)
class CalendarNameLister:
    async def __call__(self) -> Iterator[str]:
        return (
            cal.name
            for cal in caldav_principal().calendars()
        )


@dataclass(frozen=True)
class PrefixFinder:
    lister: NameLister

    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:
    finder: PrefixFinder

    @property
    def responses(self) -> dict:
        return {
            status.HTTP_200_OK: {
                "description": "Operation successful",
            },
            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