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 .. import caldav_principal, webdav_list
from ..dav_file import DavFile


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

    @property
    def responses(self) -> dict:
        return {
            status.HTTP_200_OK: {
                "description": "Operation successful",
            },
            status.HTTP_404_NOT_FOUND: {
                "description": "file not found in " +
                repr(self.finder.lister.remote_path),
                "content": None,
            },
            status.HTTP_409_CONFLICT: {
                "description": "ambiguous file name for " +
                repr(self.finder.lister.remote_path),
                "content": None,
            },
        }

    async def __call__(self, prefix: str) -> DavFile:
        file_names = list(await self.finder(prefix))

        if not (file_names):
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)

        elif len(file_names) > 1:
            raise HTTPException(status_code=status.HTTP_409_CONFLICT)

        return DavFile(f"{self.finder.lister.remote_path}/{file_names[0]}")