140 lines
3 KiB
Python
140 lines
3 KiB
Python
"""
|
|
Dependables for defining Routers.
|
|
"""
|
|
|
|
import re
|
|
from logging import getLogger
|
|
from typing import Awaitable, Callable
|
|
|
|
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__)
|
|
|
|
|
|
_RESPONSE_OK = {
|
|
status.HTTP_200_OK: {
|
|
"description": "Operation successful",
|
|
},
|
|
}
|
|
|
|
|
|
async def get_remote_path(
|
|
path_name: str,
|
|
*,
|
|
cfg: Config = Depends(get_config),
|
|
) -> str:
|
|
return getattr(cfg, path_name)
|
|
|
|
|
|
def get_lf_responses(
|
|
path: str = Depends(get_remote_path),
|
|
) -> dict:
|
|
return {
|
|
**_RESPONSE_OK,
|
|
status.HTTP_404_NOT_FOUND: {
|
|
"description": f"{path!r} not found",
|
|
"content": None,
|
|
},
|
|
}
|
|
|
|
|
|
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)
|
|
|
|
|
|
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 get_fp_responses(
|
|
src: Callable[[], Awaitable[list[str]]],
|
|
) -> dict:
|
|
return {
|
|
**_RESPONSE_OK,
|
|
status.HTTP_404_NOT_FOUND: {
|
|
"description": f"Failure in lister {src.__name__!r}",
|
|
"content": None,
|
|
},
|
|
}
|
|
|
|
|
|
async def filter_prefix(
|
|
src: Callable[[], Awaitable[list[str]]],
|
|
prefix: str = "",
|
|
) -> list[str]:
|
|
"""
|
|
Filter names from an async source `src` for names starting with a given prefix.
|
|
"""
|
|
|
|
return list(
|
|
item for item in (await src()) if item.lower().startswith(prefix.lower())
|
|
)
|
|
|
|
|
|
def get_fpu_responses() -> 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 filter_prefix_unique(
|
|
src: Callable[[str], Awaitable[list[str]]],
|
|
prefix: 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.
|
|
"""
|
|
|
|
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)
|