113 lines
3.1 KiB
Python
113 lines
3.1 KiB
Python
import logging
|
|
import re
|
|
from io import BytesIO
|
|
|
|
from asyncify import asyncify
|
|
from cache import AsyncTTL
|
|
from cache.key import KEY
|
|
from webdav3.client import Client as WebDAVclient
|
|
|
|
from .settings import SETTINGS
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
class WebDAV:
|
|
_webdav_client = WebDAVclient(
|
|
{
|
|
"webdav_hostname": SETTINGS.webdav.url,
|
|
"webdav_login": SETTINGS.webdav.username,
|
|
"webdav_password": SETTINGS.webdav.password,
|
|
"disable_check": SETTINGS.webdav.disable_check,
|
|
}
|
|
)
|
|
|
|
@classmethod
|
|
@AsyncTTL(
|
|
time_to_live=SETTINGS.webdav.cache_ttl,
|
|
maxsize=SETTINGS.webdav.cache_size,
|
|
skip_args=1,
|
|
)
|
|
async def list_files(
|
|
cls,
|
|
directory: str = "",
|
|
*,
|
|
regex: re.Pattern[str] = re.compile(""),
|
|
) -> list[str]:
|
|
"""
|
|
Liste aller Dateien im Ordner `directory`, die zur RegEx `regex` passen
|
|
"""
|
|
|
|
_logger.debug(f"list_files {directory!r}")
|
|
ls = await asyncify(cls._webdav_client.list)(directory)
|
|
|
|
return [f"{directory}/{path}" for path in ls if regex.search(path)]
|
|
|
|
@classmethod
|
|
@AsyncTTL(
|
|
time_to_live=SETTINGS.webdav.cache_ttl,
|
|
maxsize=SETTINGS.webdav.cache_size,
|
|
skip_args=1,
|
|
)
|
|
async def file_exists(cls, path: str) -> bool:
|
|
"""
|
|
`True`, wenn an Pfad `path` eine Datei existiert
|
|
"""
|
|
|
|
_logger.debug(f"file_exists {path!r}")
|
|
return await asyncify(cls._webdav_client.check)(path)
|
|
|
|
@classmethod
|
|
@(
|
|
_rb_ttl := AsyncTTL(
|
|
time_to_live=SETTINGS.webdav.cache_ttl,
|
|
maxsize=SETTINGS.webdav.cache_size,
|
|
skip_args=1,
|
|
)
|
|
)
|
|
async def read_bytes(cls, path: str) -> bytes:
|
|
"""
|
|
Datei aus Pfad `path` als bytes laden
|
|
"""
|
|
|
|
_logger.debug(f"read_bytes {path!r}")
|
|
buffer = BytesIO()
|
|
await asyncify(cls._webdav_client.resource(path).write_to)(buffer)
|
|
|
|
return buffer.read()
|
|
|
|
@classmethod
|
|
async def read_str(cls, path: str, encoding="utf-8") -> str:
|
|
"""
|
|
Datei aus Pfad `path` als string laden
|
|
"""
|
|
|
|
_logger.debug(f"read_str {path!r}")
|
|
return (await cls.read_bytes(path)).decode(encoding=encoding).strip()
|
|
|
|
@classmethod
|
|
async def write_bytes(cls, path: str, buffer: bytes) -> None:
|
|
"""
|
|
Bytes `buffer` in Datei in Pfad `path` schreiben
|
|
"""
|
|
|
|
_logger.debug(f"write_bytes {path!r}")
|
|
await asyncify(cls._webdav_client.resource(path).read_from)(buffer)
|
|
|
|
try:
|
|
# hack: zugehörigen Cache-Eintrag entfernen
|
|
# -> AsyncTTL._TTL.__contains__
|
|
del cls._rb_ttl.ttl[KEY((path,), {})]
|
|
|
|
except KeyError:
|
|
# Cache-Eintrag existierte nicht
|
|
pass
|
|
|
|
@classmethod
|
|
async def write_str(cls, path: str, content: str, encoding="utf-8") -> None:
|
|
"""
|
|
String `content` in Datei in Pfad `path` schreiben
|
|
"""
|
|
|
|
_logger.debug(f"write_str {path!r}")
|
|
await cls.write_bytes(path, content.encode(encoding=encoding))
|