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

refac: use global RP_ and LM_ objects

Browse source
This commit is contained in:
Jörn-Michael Miehe 2023-10-26 16:31:12 +02:00
parent c47773e70d
commit 44a165c53e
5 changed files with 71 additions and 144 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -7,7 +7,7 @@ from dataclasses import dataclass
from logging import getLogger
from typing import Awaitable, Callable, Generic, ParamSpec, TypeVar
from fastapi import Depends, HTTPException, status
from fastapi import Depends, HTTPException, params, status
from webdav3.exceptions import RemoteResourceNotFound
from ...core.config import get_config
@ -27,12 +27,12 @@ _RESPONSE_OK = {
Params = ParamSpec("Params")
Return = TypeVar("Return")
type _DepCallable[**Params, Return] = Callable[Params, Awaitable[Return]]
type _Dependable[**Params, Return] = Callable[Params, Awaitable[Return]]
@dataclass(slots=True, frozen=True)
class Dependable(Generic[Params, Return]):
func: _DepCallable[Params, Return]
func: _Dependable[Params, Return]
responses: dict
@ -52,6 +52,11 @@ class ListManager:
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__(
@ -59,8 +64,14 @@ class ListManager:
)
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
@ -93,7 +104,7 @@ class ListManager:
def get_remote_path(
path_name: str,
) -> _DepCallable[[], str]:
) -> _Dependable[[], str]:
async def _get_remote_path() -> str:
cfg = await get_config()
return getattr(cfg, path_name)
@ -101,8 +112,13 @@ def get_remote_path(
return _get_remote_path
def list_files(
rp: _DepCallable[[], str],
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]]:
@ -113,6 +129,11 @@ def list_files(
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)
@ -132,6 +153,18 @@ def list_files(
)
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
@ -146,71 +179,3 @@ def list_files(
# List aggregate calendar names
# """
# return list(cfg.calendar.aggregates.keys())
def filter_prefix(
src: Dependable[[], 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.func())
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.func(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,
},
},
)

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

@ -6,7 +6,6 @@ Router "file" provides:
- getting files by name prefix
"""
import re
from io import BytesIO
from logging import getLogger
@ -16,30 +15,19 @@ from magic import Magic
from ...core.dav_common import webdav_ensure_files, webdav_ensure_path
from ...core.webdav import WebDAV
from ._common import ListManager, get_remote_path, list_files
from ._common import LM_FILE, RP_FILE
_logger = getLogger(__name__)
_magic = Magic(mime=True)
router = APIRouter(prefix="/file", tags=["file"])
_rp = get_remote_path("file_dir")
_files = ListManager(
list_files(
rp=_rp,
re=re.compile(
r"[^/]$",
flags=re.IGNORECASE,
),
)
)
@router.on_event("startup")
async def start_router() -> None:
_logger.debug(f"{router.prefix} router starting.")
remote_path = await _rp()
remote_path = await RP_FILE()
if not webdav_ensure_path(remote_path):
webdav_ensure_files(
remote_path,
@ -50,32 +38,32 @@ async def start_router() -> None:
@router.get(
"/list",
responses=_files.lister.responses,
responses=LM_FILE.lister.responses,
)
async def list_all_files(
names: list[str] = Depends(_files.lister.func),
names: list[str] = Depends(LM_FILE.lister.func),
) -> list[str]:
return names
@router.get(
"/find/{prefix}",
responses=_files.filter.responses,
responses=LM_FILE.filter.responses,
)
async def find_files_by_prefix(
names: list[str] = Depends(_files.filter.func),
names: list[str] = Depends(LM_FILE.filter.func),
) -> list[str]:
return names
@router.get(
"/get/{prefix}",
responses=_files.getter.responses,
responses=LM_FILE.getter.responses,
response_class=StreamingResponse,
)
async def get_file_by_prefix(
remote_path: str = Depends(_rp),
name: str = Depends(_files.getter.func),
remote_path: str = Depends(RP_FILE),
name: str = Depends(LM_FILE.getter.func),
) -> StreamingResponse:
buffer = BytesIO(await WebDAV.read_bytes(f"{remote_path}/{name}"))

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

@ -6,7 +6,6 @@ Router "image" provides:
- getting image files in a uniform format by name prefix
"""
import re
from io import BytesIO
from logging import getLogger
@ -17,31 +16,19 @@ 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 filter_prefix, filter_prefix_unique, get_remote_path, list_files
from ._common import LM_IMAGE, RP_IMAGE
_logger = getLogger(__name__)
_PATH_NAME = "image_dir"
router = APIRouter(prefix="/image", tags=["image"])
_rp = get_remote_path(path_name=_PATH_NAME)
_ls = list_files(
rp=_rp,
re=re.compile(
r"\.(gif|jpe?g|tiff?|png|bmp)$",
flags=re.IGNORECASE,
),
)
_fp = filter_prefix(_ls)
_fpu = filter_prefix_unique(_fp)
@router.on_event("startup")
async def start_router() -> None:
_logger.debug(f"{router.prefix} router starting.")
remote_path = await _rp()
remote_path = await RP_IMAGE()
if not webdav_ensure_path(remote_path):
webdav_ensure_files(
remote_path,
@ -53,32 +40,32 @@ async def start_router() -> None:
@router.get(
"/list",
responses=_ls.responses,
responses=LM_IMAGE.lister.responses,
)
async def list_all_images(
names: list[str] = Depends(_ls.func),
names: list[str] = Depends(LM_IMAGE.lister.func),
) -> list[str]:
return names
@router.get(
"/find/{prefix}",
responses=_fp.responses,
responses=LM_IMAGE.filter.responses,
)
async def find_images_by_prefix(
names: list[str] = Depends(_fp.func),
names: list[str] = Depends(LM_IMAGE.filter.func),
) -> list[str]:
return names
@router.get(
"/get/{prefix}",
responses=_fpu.responses,
responses=LM_IMAGE.getter.responses,
response_class=StreamingResponse,
)
async def get_image_by_prefix(
remote_path: str = Depends(_rp),
name: str = Depends(_fpu.func),
remote_path: str = Depends(RP_IMAGE),
name: str = Depends(LM_IMAGE.getter.func),
) -> StreamingResponse:
cfg = await get_config()
img = Image.open(BytesIO(await WebDAV.read_bytes(f"{remote_path}/{name}")))

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

@ -7,7 +7,6 @@ Router "text" provides:
- getting text file HTML content by name prefix (using Markdown)
"""
import re
from logging import getLogger
from fastapi import APIRouter, Depends
@ -15,31 +14,19 @@ from markdown import markdown
from ...core.dav_common import webdav_ensure_files, webdav_ensure_path
from ...core.webdav import WebDAV
from ._common import filter_prefix, filter_prefix_unique, get_remote_path, list_files
from ._common import LM_TEXT, RP_TEXT
_logger = getLogger(__name__)
_PATH_NAME = "text_dir"
router = APIRouter(prefix="/text", tags=["text"])
_rp = get_remote_path(path_name=_PATH_NAME)
_ls = list_files(
rp=_rp,
re=re.compile(
r"\.(txt|md)$",
flags=re.IGNORECASE,
),
)
_fp = filter_prefix(_ls)
_fpu = filter_prefix_unique(_fp)
@router.on_event("startup")
async def start_router() -> None:
_logger.debug(f"{router.prefix} router starting.")
remote_path = await _rp()
remote_path = await RP_TEXT()
if not webdav_ensure_path(remote_path):
webdav_ensure_files(
remote_path,
@ -51,34 +38,34 @@ async def start_router() -> None:
@router.get(
"/list",
responses=_ls.responses,
responses=LM_TEXT.lister.responses,
)
async def list_all_texts(
names: list[str] = Depends(_ls.func),
names: list[str] = Depends(LM_TEXT.lister.func),
) -> list[str]:
return names
@router.get(
"/find/{prefix}",
responses=_fp.responses,
responses=LM_TEXT.filter.responses,
)
async def find_texts_by_prefix(
names: list[str] = Depends(_fp.func),
names: list[str] = Depends(LM_TEXT.filter.func),
) -> list[str]:
return names
async def _get_raw_text_by_prefix(
remote_path: str = Depends(_rp),
name: str = Depends(_fpu.func),
remote_path: str = Depends(RP_TEXT),
name: str = Depends(LM_TEXT.getter.func),
) -> str:
return await WebDAV.read_str(f"{remote_path}/{name}")
@router.get(
"/get/raw/{prefix}",
responses=_fpu.responses,
responses=LM_TEXT.getter.responses,
)
async def get_raw_text_by_prefix(
text: str = Depends(_get_raw_text_by_prefix),
@ -88,7 +75,7 @@ async def get_raw_text_by_prefix(
@router.get(
"/get/html/{prefix}",
responses=_fpu.responses,
responses=LM_TEXT.getter.responses,
)
async def get_html_by_prefix(
text: str = Depends(_get_raw_text_by_prefix),

8
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 _fpu, _rp
from ._common import LM_TEXT, RP_TEXT
_logger = getLogger(__name__)
@ -26,7 +26,7 @@ router = APIRouter(prefix="/ticker", tags=["text"])
async def start_router() -> None:
_logger.debug(f"{router.prefix} router starting.")
remote_path = await _rp()
remote_path = await RP_TEXT()
if not webdav_ensure_path(remote_path):
webdav_ensure_files(
remote_path,
@ -36,8 +36,8 @@ async def start_router() -> None:
async def get_ticker_lines() -> Iterator[str]:
cfg = await get_config()
file_name = await _fpu.func(cfg.ticker.file_name)
remote_path = await _rp()
file_name = await LM_TEXT.getter.func(cfg.ticker.file_name)
remote_path = await RP_TEXT()
ticker = await WebDAV.read_str(f"{remote_path}/{file_name}")