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

"""
Dependables for defining Routers.
"""

import re
from dataclasses import dataclass, field
from logging import getLogger
from typing import Awaitable, Callable, Generic, ParamSpec, Self, TypeVar

from fastapi import Depends, HTTPException, params, 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__)

_RESPONSE_OK = {
    status.HTTP_200_OK: {
        "description": "Operation successful",
    },
}

Params = ParamSpec("Params")
Return = TypeVar("Return")

type DependableFn[**Params, Return] = Callable[Params, Awaitable[Return]]


@dataclass(slots=True, frozen=True)
class Dependable(Generic[Params, Return]):
    func: DependableFn[Params, Return]
    responses: dict = field(default_factory=lambda: _RESPONSE_OK.copy())


@dataclass(slots=True, frozen=True)
class ListManager:
    lister: Dependable[[], list[str]]
    filter: Dependable[[str], list[str]]
    getter: Dependable[[str], str]

    @classmethod
    def from_lister(cls, lister: Dependable[[], list[str]]) -> Self:
        async def _filter_fn(
            prefix: str,
            names: list[str] = Depends(lister.func),
        ) -> list[str]:
            if isinstance(names, params.Depends):
                names = await lister.func()

            _logger.debug("filter %s from %s", repr(prefix), repr(names))

            return [item for item in names if item.lower().startswith(prefix.lower())]

        async def _getter_fn(
            prefix: str,
            names: list[str] = Depends(_filter_fn),
        ) -> str:
            if isinstance(names, params.Depends):
                names = await _filter_fn(prefix)

            _logger.debug("get %s from %s", repr(prefix), repr(names))

            match names:
                case [name]:
                    return name

                case []:
                    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)

                case _:
                    raise HTTPException(status_code=status.HTTP_409_CONFLICT)

        return cls(
            lister=lister,
            filter=Dependable(_filter_fn),
            getter=Dependable(
                func=_getter_fn,
                responses={
                    **_RESPONSE_OK,
                    status.HTTP_404_NOT_FOUND: {
                        "description": "Prefix not found",
                        "content": None,
                    },
                    status.HTTP_409_CONFLICT: {
                        "description": "Ambiguous prefix",
                        "content": None,
                    },
                },
            ),
        )

    @classmethod
    def from_lister_fn(cls, lister_fn: DependableFn[[], list[str]]) -> Self:
        return cls.from_lister(Dependable(lister_fn))


def get_remote_path(
    path_name: str,
) -> DependableFn[[], str]:
    async def _get_remote_path() -> str:
        cfg = await get_config()
        return getattr(cfg, path_name)

    return _get_remote_path


RP_FILE = get_remote_path("file_dir")
RP_IMAGE = get_remote_path("image_dir")
RP_TEXT = get_remote_path("text_dir")


def get_file_lister(
    rp: DependableFn[[], str],
    *,
    re: re.Pattern[str],
) -> Dependable[[], list[str]]:
    """
    List files in remote `path` matching the RegEx `re`
    """

    async def _list_files(
        remote_path: str = Depends(rp),
    ) -> list[str]:
        if isinstance(remote_path, params.Depends):
            remote_path = await rp()

        _logger.debug("list %s", repr(remote_path))

        try:
            return await WebDAV.list_files(remote_path, regex=re)

        except RemoteResourceNotFound:
            _logger.error("WebDAV path %s lost!", repr(remote_path))
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)

    return Dependable(
        func=_list_files,
        responses={
            **_RESPONSE_OK,
            status.HTTP_404_NOT_FOUND: {
                "description": "Remote path not found",
                "content": None,
            },
        },
    )


LM_FILE = ListManager.from_lister(
    get_file_lister(rp=RP_FILE, re=re.compile(r"[^/]$", flags=re.IGNORECASE))
)
LM_IMAGE = ListManager.from_lister(
    get_file_lister(
        rp=RP_IMAGE, re=re.compile(r"\.(gif|jpe?g|tiff?|png|bmp)$", flags=re.IGNORECASE)
    )
)
LM_TEXT = ListManager.from_lister(
    get_file_lister(rp=RP_TEXT, re=re.compile(r"\.(txt|md)$", flags=re.IGNORECASE))
)


async def list_calendar_names() -> list[str]:
    """
    List calendar names
    """

    return await CalDAV.calendars


LM_CALENDARS = ListManager.from_lister_fn(list_calendar_names)


async def list_aggregate_names(
    cfg: Config = Depends(get_config),
) -> list[str]:
    """
    List aggregate calendar names
    """

    if isinstance(cfg, params.Depends):
        cfg = await get_config()

    return list(cfg.calendar.aggregates.keys())


LM_AGGREGATES = ListManager.from_lister_fn(list_aggregate_names)