advent22/api/advent22_api/core/webdav.py

import asyncio
import re
from io import BytesIO

from cache import AsyncTTL
from cache.key import KEY
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,
        }
    )

    @classmethod
    @AsyncTTL(time_to_live=SETTINGS.webdav.cache_ttl, 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
        """

        loop = asyncio.get_running_loop()
        ls = await loop.run_in_executor(
            None,
            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, skip_args=1)
    async def file_exists(cls, path: str) -> bool:
        """
        `True`, wenn an Pfad `path` eine Datei existiert
        """

        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(
            None,
            cls._webdav_client.check,
            path,
        )

    @classmethod
    @(_rb_ttl := AsyncTTL(time_to_live=SETTINGS.webdav.cache_ttl, skip_args=1))
    async def read_bytes(cls, path: str) -> bytes:
        """
        Datei aus Pfad `path` als bytes laden
        """

        buffer = BytesIO()
        loop = asyncio.get_running_loop()
        await loop.run_in_executor(
            None,
            cls._webdav_client.resource(path).write_to,
            buffer,
        )
        buffer.seek(0)

        return buffer.read()

    @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_bytes(cls, path: str, buffer: bytes) -> None:
        """
        Bytes `buffer` in Datei in Pfad `path` schreiben
        """

        loop = asyncio.get_running_loop()
        await loop.run_in_executor(
            None,
            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
        """

        await cls.write_bytes(path, content.encode(encoding=encoding))