ovdashboard/api/ovdashboard_api/routers/v1/_common.py

"""
Dependables for defining Routers.
"""

import re
from dataclasses import dataclass
from logging import getLogger
from typing import Iterator, Protocol

from fastapi import Depends, HTTPException, status
from webdav3.exceptions import RemoteResourceNotFound

from ...core.caldav import CalDAV
from ...core.config import Config, 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)


async def get_remote_path(
    path_name: str,
    *,
    cfg: Config = Depends(get_config),
) -> str:
    return getattr(cfg, path_name)


async def list_files(
    re: re.Pattern[str],
    *,
    path: str = Depends(get_remote_path),
) -> list[str]:
    """
    List files in remote `path` matching the RegEx `re`
    """
    try:
        return await WebDAV.list_files(path, regex=re)

    except RemoteResourceNotFound:
        _logger.error(
            "WebDAV path %s lost!",
            repr(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)


async def list_calendar_names() -> list[str]:
    """
    List calendar names
    """
    return 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())


async def list_aggregate_names(
    cfg: Config = Depends(get_config),
) -> list[str]:
    """
    List aggregate calendar names
    """
    return list(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