ovdashboard/api/ovdashboard_api/core/webdav.py

import logging
import re
from io import BytesIO

from asyncify import asyncify
from cache import AsyncTTL
from cache.key import KEY
from webdav3.client import Client as WebDAVclient

from .settings import SETTINGS

_logger = logging.getLogger(__name__)


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,
        }
    )

    @classmethod
    @AsyncTTL(
        time_to_live=SETTINGS.webdav.cache_ttl,
        maxsize=SETTINGS.webdav.cache_size,
        skip_args=1,
    )
    async def list_files(
        cls,
        directory: str = "",
        *,
        regex: re.Pattern[str] = re.compile(""),
    ) -> list[str]:
        """
        List files in directory `directory` matching RegEx `regex`
        """

        _logger.debug(f"list_files {directory!r}")
        ls = await asyncify(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,
        maxsize=SETTINGS.webdav.cache_size,
        skip_args=1,
    )
    async def exists(cls, path: str) -> bool:
        """
        `True` iff there is a WebDAV resource at `path`
        """

        _logger.debug(f"file_exists {path!r}")
        return await asyncify(cls._webdav_client.check)(path)

    @classmethod
    @(
        _rb_ttl := AsyncTTL(
            time_to_live=SETTINGS.webdav.cache_ttl,
            maxsize=SETTINGS.webdav.cache_size,
            skip_args=1,
        )
    )
    async def read_bytes(cls, path: str) -> bytes:
        """
        Load WebDAV file from `path` as bytes
        """

        _logger.debug(f"read_bytes {path!r}")
        buffer = BytesIO()
        await asyncify(cls._webdav_client.resource(path).write_to)(buffer)

        return buffer.read()

    @classmethod
    async def read_str(cls, path: str, encoding="utf-8") -> str:
        """
        Load WebDAV file from `path` as string
        """

        _logger.debug(f"read_str {path!r}")
        return (await cls.read_bytes(path)).decode(encoding=encoding).strip()

    @classmethod
    async def write_bytes(cls, path: str, buffer: bytes) -> None:
        """
        Write bytes from `buffer` into WebDAV file at `path`
        """

        _logger.debug(f"write_bytes {path!r}")
        await asyncify(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:
        """
        Write string from `content` into WebDAV file at `path`
        """

        _logger.debug(f"write_str {path!r}")
        await cls.write_bytes(path, content.encode(encoding=encoding))