import re from io import BytesIO from cache import AsyncTTL from webdav3.client import Client as WebDAVclient from ..settings import SETTINGS 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, } ) @AsyncTTL(time_to_live=SETTINGS.cache_ttl) @classmethod 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 """ ls = cls._webdav_client.list(directory) return [f"{directory}/{path}" for path in ls if regex.search(path)] @AsyncTTL(time_to_live=SETTINGS.cache_ttl) @classmethod async def file_exists(cls, path: str) -> bool: """ `True`, wenn an Pfad `path` eine Datei existiert """ return cls._webdav_client.check(path) @AsyncTTL(time_to_live=SETTINGS.cache_ttl) @classmethod async def read_bytes(cls, path: str) -> bytes: """ Datei aus Pfad `path` als bytes laden """ buffer = BytesIO() cls._webdav_client.resource(path).write_to(buffer) return buffer.read() @AsyncTTL(time_to_live=SETTINGS.cache_ttl) @classmethod async def read_str(cls, path: str, encoding="utf-8") -> str: """ Datei aus Pfad `path` als string laden """ return (await cls.read_bytes(path)).decode(encoding=encoding).strip() @classmethod async def write_buffer(cls, path: str, buffer: BytesIO) -> None: """ Puffer `buffer` in Datei in Pfad `path` schreiben """ cls._webdav_client.resource(path).read_from(buffer) @classmethod async def write_str(cls, path: str, content: str, encoding="utf-8") -> None: """ String `content` in Datei in Pfad `path` schreiben """ buffer = BytesIO(content.encode(encoding=encoding)) buffer.seek(0) await cls.write_buffer(path, buffer)