141 lines
3.3 KiB
Python
141 lines
3.3 KiB
Python
"""
|
|
Dependables for defining Routers.
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
import tomllib
|
|
|
|
import tomli_w
|
|
from fastapi import Depends, HTTPException, params, status
|
|
from webdav3.exceptions import RemoteResourceNotFound
|
|
|
|
from ...core.config import Config
|
|
from ...core.dav.caldav import CalDAV
|
|
from ...core.dav.webdav import WebDAV
|
|
from ...core.settings import SETTINGS
|
|
from ._list_manager import Dependable, DependableFn, ListManager
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
_RESPONSE_OK = {
|
|
status.HTTP_200_OK: {
|
|
"description": "Operation successful",
|
|
},
|
|
}
|
|
|
|
|
|
async def get_config() -> Config:
|
|
"""
|
|
Load the configuration instance from the server using `TOML`.
|
|
"""
|
|
|
|
try:
|
|
cfg_str = await WebDAV.read_str(SETTINGS.webdav.config_filename)
|
|
cfg = Config.model_validate(tomllib.loads(cfg_str))
|
|
|
|
except RemoteResourceNotFound:
|
|
_logger.warning(
|
|
f"Config file {SETTINGS.webdav.config_filename!r} not found, creating ..."
|
|
)
|
|
|
|
cfg = Config()
|
|
cfg.calendar.aggregates["All Events"] = list(await CalDAV.calendars)
|
|
|
|
await WebDAV.write_str(
|
|
SETTINGS.webdav.config_filename,
|
|
tomli_w.dumps(cfg.model_dump()),
|
|
)
|
|
|
|
return cfg
|
|
|
|
|
|
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_CALENDAR = 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_AGGREGATE = ListManager.from_lister_fn(list_aggregate_names)
|