advent22/api/advent22_api/core/dav/helpers.py

from json import JSONDecodeError
from typing import Callable, Hashable

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


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:
        if isinstance(s, bytes):
            return s

        else:
            return super()._serialize(s)

    def _deserialize(self, s: ResponseT):
        try:
            return super()._deserialize(s)

        except (UnicodeDecodeError, JSONDecodeError):
            assert isinstance(s, bytes)
            return s