OEKZident.de/ovdashboard
2
0
Fork 0
Code Releases 1 Packages Activity

wip: functionify routers._common

Browse source
This commit is contained in:
Jörn-Michael Miehe 2023-10-23 23:32:25 +02:00
parent e6509d85fc
commit 78b1359603
5 changed files with 210 additions and 178 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

163
api/ovdashboard_api/routers/v1/_common.py
View file

@ -3,16 +3,19 @@ Dependables for defining Routers.
"""
import re
from dataclasses import dataclass
from logging import getLogger
from typing import Awaitable, Callable
from typing import Awaitable, Callable, ParamSpec, TypeVar
from fastapi import Depends, HTTPException, status
from fastapi import HTTPException, status
from webdav3.exceptions import RemoteResourceNotFound
from ...core.caldav import CalDAV
from ...core.config import Config, get_config
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__)
@ -22,35 +25,42 @@ _RESPONSE_OK = {
},
}
Params = ParamSpec("Params")
Return = TypeVar("Return")
async def get_remote_path(
@dataclass(slots=True, frozen=True)
class Dependable[**Params, Return]:
func: Callable[Params, Return]
responses: dict
def get_remote_path(
path_name: str,
*,
cfg: Config = Depends(get_config),
) -> str:
) -> Dependable[[], Awaitable[str]]:
async def _get_remote_path() -> str:
cfg = await get_config()
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,
},
}
return Dependable(
func=_get_remote_path,
responses={**_RESPONSE_OK},
)
async def list_files(
re: re.Pattern[str],
def list_files(
*,
path: str = Depends(get_remote_path),
) -> list[str]:
path_name: str,
re: re.Pattern[str],
) -> Dependable[[], Awaitable[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)
@ -61,72 +71,72 @@ async def list_files(
)
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
async def list_calendar_names() -> list[str]:
"""
List calendar names
"""
return await CalDAV.calendars
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_aggregate_names(
cfg: Config = Depends(get_config),
) -> list[str]:
"""
List aggregate calendar names
"""
return list(cfg.calendar.aggregates.keys())
# async def list_calendar_names() -> list[str]:
# """
# List calendar names
# """
# return await CalDAV.calendars
def get_fp_responses(
# 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: Callable[[], Awaitable[list[str]]],
) -> dict:
return {
) -> Dependable[[str], Awaitable[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.__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(
def filter_prefix_unique(
src: Callable[[str], Awaitable[list[str]]],
prefix: str = "",
) -> str:
) -> Dependable[[str], Awaitable[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:
@ -138,3 +148,18 @@ async def filter_prefix_unique(
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,
},
},
)

43
api/ovdashboard_api/routers/v1/file.py
View file

@ -9,7 +9,6 @@ Router "file" provides:
import re
from io import BytesIO
from logging import getLogger
from typing import Iterator
from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse
@ -17,32 +16,35 @@ from magic import Magic
from ...core.dav_common import webdav_ensure_files, webdav_ensure_path
from ...core.webdav import WebDAV
from ._common import FileNameLister, PrefixFinder, PrefixUnique
from ._common import filter_prefix, filter_prefix_unique, get_remote_path, list_files
_logger = getLogger(__name__)
_magic = Magic(mime=True)
_PATH_NAME = "file_dir"
router = APIRouter(prefix="/file", tags=["file"])
file_lister = FileNameLister(
path_name="file_dir",
_ls = list_files(
path_name=_PATH_NAME,
re=re.compile(
r"[^/]$",
flags=re.IGNORECASE,
),
)
file_finder = PrefixFinder(file_lister)
file_unique = PrefixUnique(file_finder)
_rp = get_remote_path(path_name=_PATH_NAME)
_fp = filter_prefix(_ls.func)
_fpu = filter_prefix_unique(_fp.func)
@router.on_event("startup")
async def start_router() -> None:
_logger.debug(f"{router.prefix} router starting.")
if not webdav_ensure_path(await file_lister.remote_path):
remote_path = await _rp.func()
if not webdav_ensure_path(remote_path):
webdav_ensure_files(
await file_lister.remote_path,
remote_path,
"logo.svg",
"thw.svg",
)
@ -51,35 +53,36 @@ async def start_router() -> None:
@router.get(
"/list",
response_model=list[str],
responses=file_lister.responses,
responses=_ls.responses,
)
async def list_files(
names: Iterator[str] = Depends(file_lister),
async def list_all_files(
names: list[str] = Depends(_ls.func),
) -> list[str]:
return list(names)
return names
@router.get(
"/find/{prefix}",
response_model=list[str],
responses=file_finder.responses,
responses=_fp.responses,
)
async def find_files(
names: Iterator[str] = Depends(file_finder),
async def find_files_by_prefix(
names: list[str] = Depends(_fp.func),
) -> list[str]:
return list(names)
return names
@router.get(
"/get/{prefix}",
response_class=StreamingResponse,
responses=file_unique.responses,
responses=_fpu.responses,
)
async def get_file(
async def get_file_by_prefix(
prefix: str,
name: str = Depends(file_unique),
remote_path: str = Depends(_rp.func),
name: str = Depends(_fpu.func),
) -> StreamingResponse:
buffer = BytesIO(await WebDAV.read_bytes(f"{await file_lister.remote_path}/{name}"))
buffer = BytesIO(await WebDAV.read_bytes(f"{remote_path}/{name}"))
mime = _magic.from_buffer(buffer.read(2048))
buffer.seek(0)

45
api/ovdashboard_api/routers/v1/image.py
View file

@ -9,7 +9,6 @@ Router "image" provides:
import re
from io import BytesIO
from logging import getLogger
from typing import Iterator
from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse
@ -18,31 +17,34 @@ from PIL import Image
from ...core.config import Config, ImageUIConfig, get_config
from ...core.dav_common import webdav_ensure_files, webdav_ensure_path
from ...core.webdav import WebDAV
from ._common import FileNameLister, PrefixFinder, PrefixUnique
from ._common import filter_prefix, filter_prefix_unique, get_remote_path, list_files
_logger = getLogger(__name__)
_PATH_NAME = "image_dir"
router = APIRouter(prefix="/image", tags=["image"])
image_lister = FileNameLister(
path_name="image_dir",
_ls = list_files(
path_name=_PATH_NAME,
re=re.compile(
r"\.(gif|jpe?g|tiff?|png|bmp)$",
flags=re.IGNORECASE,
),
)
image_finder = PrefixFinder(image_lister)
image_unique = PrefixUnique(image_finder)
_rp = get_remote_path(path_name=_PATH_NAME)
_fp = filter_prefix(_ls.func)
_fpu = filter_prefix_unique(_fp.func)
@router.on_event("startup")
async def start_router() -> None:
_logger.debug(f"{router.prefix} router starting.")
if not webdav_ensure_path(await image_lister.remote_path):
remote_path = await _rp.func()
if not webdav_ensure_path(remote_path):
webdav_ensure_files(
await image_lister.remote_path,
remote_path,
"img1.jpg",
"img2.jpg",
"img3.jpg",
@ -52,38 +54,37 @@ async def start_router() -> None:
@router.get(
"/list",
response_model=list[str],
responses=image_lister.responses,
responses=_ls.responses,
)
async def list_images(
names: Iterator[str] = Depends(image_lister),
async def list_all_images(
names: list[str] = Depends(_ls.func),
) -> list[str]:
return list(names)
return names
@router.get(
"/find/{prefix}",
response_model=list[str],
responses=image_finder.responses,
responses=_fp.responses,
)
async def find_images(
names: Iterator[str] = Depends(image_finder),
async def find_images_by_prefix(
names: list[str] = Depends(_fp.func),
) -> list[str]:
return list(names)
return names
@router.get(
"/get/{prefix}",
response_class=StreamingResponse,
responses=image_unique.responses,
responses=_fpu.responses,
)
async def get_image(
async def get_image_by_prefix(
prefix: str,
name: str = Depends(image_unique),
remote_path: str = Depends(_rp.func),
name: str = Depends(_fpu.func),
) -> StreamingResponse:
cfg = await get_config()
img = Image.open(
BytesIO(await WebDAV.read_bytes(f"{await image_lister.remote_path}/{name}"))
)
img = Image.open(BytesIO(await WebDAV.read_bytes(f"{remote_path}/{name}")))
img_buffer = BytesIO()
img.save(img_buffer, **cfg.image.save_params)

70
api/ovdashboard_api/routers/v1/text.py
View file

@ -9,39 +9,40 @@ Router "text" provides:
import re
from logging import getLogger
from typing import Iterator
from fastapi import APIRouter, Depends
from markdown import markdown
from ...core.dav_common import webdav_ensure_files, webdav_ensure_path
from ...core.webdav import WebDAV
from ._common import FileNameLister, PrefixFinder, PrefixUnique
from ._common import filter_prefix, filter_prefix_unique, get_remote_path, list_files
_logger = getLogger(__name__)
_PATH_NAME = "text_dir"
router = APIRouter(prefix="/text", tags=["text"])
text_lister = FileNameLister(
path_name="text_dir",
_ls = list_files(
path_name=_PATH_NAME,
re=re.compile(
r"\.(txt|md)$",
flags=re.IGNORECASE,
),
)
text_finder = PrefixFinder(text_lister)
text_unique = PrefixUnique(text_finder)
_rp = get_remote_path(path_name=_PATH_NAME)
_fp = filter_prefix(_ls.func)
_fpu = filter_prefix_unique(_fp.func)
@router.on_event("startup")
async def start_router() -> None:
_logger.debug(f"{router.prefix} router starting.")
webdav_ensure_path(await text_lister.remote_path)
remote_path = await _rp.func()
if not webdav_ensure_path(remote_path):
webdav_ensure_files(
await text_lister.remote_path,
remote_path,
"message.txt",
"title.txt",
"ticker.txt",
@ -51,48 +52,49 @@ async def start_router() -> None:
@router.get(
"/list",
response_model=list[str],
responses=text_lister.responses,
responses=_ls.responses,
)
async def list_texts(
names: Iterator[str] = Depends(text_lister),
async def list_all_texts(
names: list[str] = Depends(_ls.func),
) -> list[str]:
return list(names)
return names
@router.get(
"/find/{prefix}",
response_model=list[str],
responses=text_finder.responses,
responses=_fp.responses,
)
async def find_texts(
names: Iterator[str] = Depends(text_finder),
async def find_texts_by_prefix(
names: list[str] = Depends(_fp.func),
) -> list[str]:
return list(names)
return names
async def get_text_content(
name: str = Depends(text_unique),
async def _get_raw_text_by_prefix(
remote_path: str = Depends(_rp.func),
name: str = Depends(_fpu.func),
) -> str:
return await WebDAV.read_str(f"{await text_lister.remote_path}/{name}")
@router.get(
"/get/html/{prefix}",
response_model=str,
responses=text_unique.responses,
)
async def get_text(
text: str = Depends(get_text_content),
) -> str:
return markdown(text)
return await WebDAV.read_str(f"{remote_path}/{name}")
@router.get(
"/get/raw/{prefix}",
response_model=str,
responses=text_unique.responses,
responses=_fpu.responses,
)
async def get_raw_text(
text: str = Depends(get_text_content),
async def get_raw_text_by_prefix(
text: str = Depends(_get_raw_text_by_prefix),
) -> str:
return text
@router.get(
"/get/html/{prefix}",
response_model=str,
responses=_fpu.responses,
)
async def get_html_by_prefix(
text: str = Depends(_get_raw_text_by_prefix),
) -> str:
return markdown(text)

13
api/ovdashboard_api/routers/v1/ticker.py
View file

@ -15,7 +15,7 @@ from markdown import markdown
from ...core.config import Config, TickerUIConfig, get_config
from ...core.dav_common import webdav_ensure_files, webdav_ensure_path
from ...core.webdav import WebDAV
from .text import text_lister, text_unique
from .text import _fpu, _rp
_logger = getLogger(__name__)
@ -26,19 +26,20 @@ router = APIRouter(prefix="/ticker", tags=["text"])
async def start_router() -> None:
_logger.debug(f"{router.prefix} router starting.")
webdav_ensure_path(await text_lister.remote_path)
remote_path = await _rp.func()
if not webdav_ensure_path(remote_path):
webdav_ensure_files(
await text_lister.remote_path,
remote_path,
"ticker.txt",
)
async def get_ticker_lines() -> Iterator[str]:
cfg = await get_config()
file_name = await text_unique(cfg.ticker.file_name)
file_name = await _fpu.func(cfg.ticker.file_name)
remote_path = await _rp.func()
ticker = await WebDAV.read_str(f"{await text_lister.remote_path}/{file_name}")
ticker = await WebDAV.read_str(f"{remote_path}/{file_name}")
return (line.strip() for line in ticker.split("\n") if line.strip())