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

import pickle
from typing import Callable, Hashable

import requests
from cachetools.keys import hashkey
from CacheToolsUtils import RedisCache as __RedisCache
from redis import Redis
from redis.commands.core import ResponseT
from redis.typing import EncodableT
from webdav3.client import Client as __WebDAVclient

from ..settings import SETTINGS


def davkey(
    name: str,
    slice: slice = slice(1, None),
) -> Callable[..., tuple[Hashable, ...]]:
    def func(*args, **kwargs) -> tuple[Hashable, ...]:
        """Return a cache key for use with cached methods."""

        key = hashkey(name, *args[slice], **kwargs)
        return hashkey(*(str(key_item) for key_item in key))

    return func


class WebDAVclient(__WebDAVclient):
    def execute_request(
        self,
        action,
        path,
        data=None,
        headers_ext=None,
    ) -> requests.Response:
        res = super().execute_request(action, path, data, headers_ext)

        # the "Content-Length" header can randomly be missing on txt files,
        # this should fix that (probably serverside bug)
        if action == "download" and "Content-Length" not in res.headers:
            res.headers["Content-Length"] = str(len(res.text))

        return res


class RedisCache(__RedisCache):
    """
    Redis handles <bytes>, so ...
    """

    def _serialize(self, s) -> EncodableT:
        return pickle.dumps(s)

    def _deserialize(self, s: ResponseT):
        assert isinstance(s, bytes)
        return pickle.loads(s)


REDIS = Redis(
    host=SETTINGS.redis.host,
    port=SETTINGS.redis.port,
    db=SETTINGS.redis.db,
    protocol=SETTINGS.redis.protocol,
)