181 lines
4.7 KiB
Python
181 lines
4.7 KiB
Python
"""
|
|
Dependables for defining Routers.
|
|
"""
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from logging import getLogger
|
|
from typing import Awaitable, Callable, Generic, ParamSpec, TypeVar
|
|
|
|
from fastapi import Depends, HTTPException, params, 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")
|
|
|
|
type _Dependable[**Params, Return] = Callable[Params, Awaitable[Return]]
|
|
|
|
|
|
@dataclass(slots=True, frozen=True)
|
|
class Dependable(Generic[Params, Return]):
|
|
func: _Dependable[Params, Return]
|
|
responses: dict
|
|
|
|
|
|
@dataclass(init=False, slots=True, frozen=True)
|
|
class ListManager:
|
|
lister: Dependable[[], list[str]]
|
|
filter: Dependable[[str], list[str]]
|
|
getter: Dependable[[str], str]
|
|
|
|
def __init__(
|
|
self,
|
|
lister: Dependable[[], list[str]],
|
|
) -> None:
|
|
object.__setattr__(self, "lister", lister)
|
|
|
|
async def _filter(
|
|
prefix: str,
|
|
names: list[str] = Depends(self.lister.func),
|
|
) -> list[str]:
|
|
if isinstance(names, params.Depends):
|
|
names = await self.lister.func()
|
|
|
|
_logger.debug("filter %s from %s", repr(prefix), repr(names))
|
|
|
|
return [item for item in names if item.lower().startswith(prefix.lower())]
|
|
|
|
object.__setattr__(
|
|
self, "filter", Dependable(func=_filter, responses=_RESPONSE_OK)
|
|
)
|
|
|
|
async def _getter(
|
|
prefix: str,
|
|
names: list[str] = Depends(self.filter.func),
|
|
) -> str:
|
|
if isinstance(names, params.Depends):
|
|
names = await self.filter.func(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)
|
|
|
|
object.__setattr__(
|
|
self,
|
|
"getter",
|
|
Dependable(
|
|
func=_getter,
|
|
responses={
|
|
**_RESPONSE_OK,
|
|
status.HTTP_404_NOT_FOUND: {
|
|
"description": "Prefix not found",
|
|
"content": None,
|
|
},
|
|
status.HTTP_409_CONFLICT: {
|
|
"description": "Ambiguous prefix",
|
|
"content": None,
|
|
},
|
|
},
|
|
),
|
|
)
|
|
|
|
|
|
def get_remote_path(
|
|
path_name: str,
|
|
) -> _Dependable[[], 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: _Dependable[[], 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(
|
|
get_file_lister(rp=RP_FILE, re=re.compile(r"[^/]$", flags=re.IGNORECASE))
|
|
)
|
|
LM_IMAGE = ListManager(
|
|
get_file_lister(
|
|
rp=RP_IMAGE, re=re.compile(r"\.(gif|jpe?g|tiff?|png|bmp)$", flags=re.IGNORECASE)
|
|
)
|
|
)
|
|
LM_TEXT = ListManager(
|
|
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
|
|
|
|
|
|
# async def list_aggregate_names(
|
|
# cfg: Config = Depends(get_config),
|
|
# ) -> list[str]:
|
|
# """
|
|
# List aggregate calendar names
|
|
# """
|
|
# return list(cfg.calendar.aggregates.keys())
|