170 lines
4 KiB
Python
170 lines
4 KiB
Python
"""
|
|
Dependables for defining Routers.
|
|
"""
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from logging import getLogger
|
|
from typing import Awaitable, Callable, ParamSpec, TypeVar
|
|
|
|
from fastapi import HTTPException, status
|
|
from webdav3.exceptions import RemoteResourceNotFound
|
|
|
|
from ...core.config import get_config
|
|
from ...core.webdav import WebDAV
|
|
|
|
# from ...core.caldav import CalDAV
|
|
# from ...core.config import Config, get_config
|
|
|
|
_logger = getLogger(__name__)
|
|
|
|
_RESPONSE_OK = {
|
|
status.HTTP_200_OK: {
|
|
"description": "Operation successful",
|
|
},
|
|
}
|
|
|
|
Params = ParamSpec("Params")
|
|
Return = TypeVar("Return")
|
|
|
|
|
|
@dataclass(slots=True, frozen=True)
|
|
class Dependable[**Params, Return]:
|
|
func: Callable[Params, Awaitable[Return]]
|
|
responses: dict
|
|
|
|
async def __call__(self, *args: Params.args, **kwds: Params.kwargs) -> Return:
|
|
return await self.func(*args, **kwds)
|
|
|
|
|
|
type _NDependable[Return] = Dependable[[], Return]
|
|
|
|
|
|
def get_remote_path(
|
|
path_name: str,
|
|
) -> _NDependable[str]:
|
|
async def _get_remote_path() -> str:
|
|
cfg = await get_config()
|
|
return getattr(cfg, path_name)
|
|
|
|
return Dependable(
|
|
func=_get_remote_path,
|
|
responses={**_RESPONSE_OK},
|
|
)
|
|
|
|
|
|
def list_files(
|
|
*,
|
|
path_name: str,
|
|
re: re.Pattern[str],
|
|
) -> _NDependable[list[str]]:
|
|
"""
|
|
List files in remote `path` matching the RegEx `re`
|
|
"""
|
|
|
|
async def _list_files() -> list[str]:
|
|
cfg = await get_config()
|
|
path = getattr(cfg, path_name)
|
|
|
|
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)
|
|
|
|
return Dependable(
|
|
func=_list_files,
|
|
responses={
|
|
**_RESPONSE_OK,
|
|
status.HTTP_404_NOT_FOUND: {
|
|
"description": f"{path_name!r} not found",
|
|
"content": None,
|
|
},
|
|
},
|
|
)
|
|
|
|
|
|
# async def list_calendar_names() -> list[str]:
|
|
# """
|
|
# List calendar names
|
|
# """
|
|
# return await CalDAV.calendars
|
|
|
|
|
|
# async def list_aggregate_names(
|
|
# cfg: Config = Depends(get_config),
|
|
# ) -> list[str]:
|
|
# """
|
|
# List aggregate calendar names
|
|
# """
|
|
# return list(cfg.calendar.aggregates.keys())
|
|
|
|
|
|
def filter_prefix(
|
|
src: _NDependable[list[str]],
|
|
) -> Dependable[[str], list[str]]:
|
|
"""
|
|
Filter names from an async source `src` for names starting with a given prefix.
|
|
"""
|
|
|
|
async def _filter_prefix(
|
|
prefix: str,
|
|
) -> list[str]:
|
|
return list(
|
|
item for item in (await src()) if item.lower().startswith(prefix.lower())
|
|
)
|
|
|
|
return Dependable(
|
|
func=_filter_prefix,
|
|
responses={
|
|
**_RESPONSE_OK,
|
|
status.HTTP_404_NOT_FOUND: {
|
|
"description": f"Failure in lister {src.__class__.__name__!r}",
|
|
"content": None,
|
|
},
|
|
},
|
|
)
|
|
|
|
|
|
def filter_prefix_unique(
|
|
src: Dependable[[str], list[str]],
|
|
) -> Dependable[[str], str]:
|
|
"""
|
|
Determines if a given prefix is unique in the list produced by the async source `src`.
|
|
|
|
On success, produces the unique name with that prefix. Otherwise, throws a HTTPException.
|
|
"""
|
|
|
|
async def _filter_prefix_unique(
|
|
prefix: str,
|
|
) -> str:
|
|
names = await src(prefix)
|
|
|
|
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 Dependable(
|
|
func=_filter_prefix_unique,
|
|
responses={
|
|
**_RESPONSE_OK,
|
|
status.HTTP_404_NOT_FOUND: {
|
|
"description": "Prefix not found",
|
|
"content": None,
|
|
},
|
|
status.HTTP_409_CONFLICT: {
|
|
"description": "Ambiguous prefix",
|
|
"content": None,
|
|
},
|
|
},
|
|
)
|