ovdashboard/api/ovdashboard_api/core/dav/webdav.py

import logging
import re
from io import BytesIO

from asyncify import asyncify
from cachetools import cachedmethod

from ..settings import SETTINGS
from .helpers import REDIS, RedisCache, WebDAVclient, davkey

_logger = logging.getLogger(__name__)


class WebDAV:
    _webdav_client = WebDAVclient(
        {
            "webdav_hostname": SETTINGS.webdav.url,
            "webdav_login": SETTINGS.webdav.username,
            "webdav_password": SETTINGS.webdav.password,
        }
    )

    _cache = RedisCache(
        cache=REDIS,
        ttl=SETTINGS.webdav.cache_ttl,
    )

    @classmethod
    @asyncify
    @cachedmethod(cache=lambda cls: cls._cache, key=davkey("list_files"))
    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 = cls._webdav_client.list(directory)

        return [path for path in ls if regex.search(path)]

    @classmethod
    @asyncify
    @cachedmethod(cache=lambda cls: cls._cache, key=davkey("exists"))
    def exists(cls, path: str) -> bool:
        """
        `True` iff there is a WebDAV resource at `path`
        """

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

    @classmethod
    @asyncify
    @cachedmethod(cache=lambda cls: cls._cache, key=davkey("read_bytes"))
    def read_bytes(cls, path: str) -> bytes:
        """
        Load WebDAV file from `path` as bytes
        """

        _logger.debug(f"read_bytes {path!r}")
        buffer = BytesIO()
        cls._webdav_client.download_from(buffer, path)
        buffer.seek(0)

        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
    @asyncify
    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}")
        cls._webdav_client.upload_to(buffer, path)

        # invalidate cache entry
        # explicit slice as there is no "cls" argument
        del cls._cache[davkey("read_bytes", slice(0, None))(path)]

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