94 lines
2.5 KiB
Python
94 lines
2.5 KiB
Python
import re
|
|
from contextlib import contextmanager
|
|
from io import BytesIO, TextIOWrapper
|
|
from typing import ContextManager, Iterator
|
|
|
|
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_buffer(cls, path: str) -> ContextManager[BytesIO]:
|
|
"""
|
|
Datei aus Pfad `path` in einen `BytesIO` Puffer laden
|
|
"""
|
|
|
|
buffer = BytesIO()
|
|
cls._webdav_client.resource(path).write_to(buffer)
|
|
|
|
@contextmanager
|
|
def ctx() -> Iterator[BytesIO]:
|
|
buffer.seek(0)
|
|
yield buffer
|
|
|
|
return ctx()
|
|
|
|
@AsyncTTL(time_to_live=SETTINGS.cache_ttl)
|
|
@classmethod
|
|
async def read_text(cls, path: str, encoding="utf-8") -> str:
|
|
"""
|
|
Datei aus Pfad `path` als string zurückgeben
|
|
"""
|
|
|
|
with await cls.read_buffer(path) as buffer:
|
|
tio = TextIOWrapper(buffer, encoding=encoding)
|
|
tio.seek(0)
|
|
|
|
return tio.read().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_text(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)
|