advent22/api/advent22_api/core/webdav.py

80 lines
2.1 KiB
Python
Raw Normal View History

2023-09-08 02:45:00 +00:00
import re
2023-09-08 16:19:26 +00:00
from io import BytesIO
2023-09-08 02:45:00 +00:00
from cache import AsyncTTL
from webdav3.client import Client as WebDAVclient
2023-09-08 18:17:18 +00:00
from .settings import SETTINGS
2023-09-08 02:45:00 +00:00
class WebDAV:
_webdav_client = WebDAVclient(
{
"webdav_hostname": SETTINGS.webdav.url,
"webdav_login": SETTINGS.webdav.username,
"webdav_password": SETTINGS.webdav.password,
}
)
@classmethod
2023-09-11 02:37:08 +00:00
@AsyncTTL(time_to_live=SETTINGS.webdav.cache_ttl, skip_args=1)
2023-09-08 02:45:00 +00:00
async def list_files(
cls,
directory: str = "",
2023-09-08 18:17:18 +00:00
*,
2023-09-08 02:45:00 +00:00
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)]
@classmethod
2023-09-11 02:37:08 +00:00
@AsyncTTL(time_to_live=SETTINGS.webdav.cache_ttl, skip_args=1)
2023-09-08 02:45:00 +00:00
async def file_exists(cls, path: str) -> bool:
"""
`True`, wenn an Pfad `path` eine Datei existiert
"""
return cls._webdav_client.check(path)
@classmethod
2023-09-11 02:37:08 +00:00
@AsyncTTL(time_to_live=SETTINGS.webdav.cache_ttl, skip_args=1)
2023-09-08 16:19:26 +00:00
async def read_bytes(cls, path: str) -> bytes:
2023-09-08 02:45:00 +00:00
"""
2023-09-08 16:19:26 +00:00
Datei aus Pfad `path` als bytes laden
2023-09-08 02:45:00 +00:00
"""
2023-09-08 16:19:26 +00:00
buffer = BytesIO()
cls._webdav_client.resource(path).write_to(buffer)
buffer.seek(0)
2023-09-08 16:19:26 +00:00
return buffer.read()
2023-09-08 02:45:00 +00:00
@classmethod
2023-09-08 16:19:26 +00:00
async def read_str(cls, path: str, encoding="utf-8") -> str:
2023-09-08 02:45:00 +00:00
"""
2023-09-08 16:19:26 +00:00
Datei aus Pfad `path` als string laden
2023-09-08 02:45:00 +00:00
"""
2023-09-08 16:19:26 +00:00
return (await cls.read_bytes(path)).decode(encoding=encoding).strip()
2023-09-08 02:45:00 +00:00
@classmethod
async def write_bytes(cls, path: str, buffer: bytes) -> None:
2023-09-08 02:45:00 +00:00
"""
Bytes `buffer` in Datei in Pfad `path` schreiben
2023-09-08 02:45:00 +00:00
"""
cls._webdav_client.resource(path).read_from(buffer)
@classmethod
2023-09-08 16:19:26 +00:00
async def write_str(cls, path: str, content: str, encoding="utf-8") -> None:
2023-09-08 02:45:00 +00:00
"""
String `content` in Datei in Pfad `path` schreiben
"""
await cls.write_bytes(path, content.encode(encoding=encoding))