advent22/api/advent22_api/core/webdav.py

import re
from contextlib import asynccontextmanager
from io import BytesIO, TextIOWrapper
from typing import AsyncContextManager, AsyncIterator

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)

    @classmethod
    async def read_buffer(cls, path: str) -> AsyncContextManager[BytesIO]:
        """
        Datei aus Pfad `path` in einen `BytesIO` Puffer laden
        """

        @AsyncTTL(time_to_live=SETTINGS.cache_ttl)
        async def inner() -> BytesIO:
            buffer = BytesIO()
            cls._webdav_client.resource(path).write_to(buffer)
            return buffer

        @asynccontextmanager
        async def ctx() -> AsyncIterator[BytesIO]:
            buffer = await inner()
            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
        """

        async 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)