advent22/api/advent22_api/core/webdav.py

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)