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

from itertools import chain
from json import JSONDecodeError
from typing import Any, Callable

import requests
from CacheToolsUtils import RedisCache as __RedisCache
from redis.typing import EncodableT, ResponseT
from webdav3.client import Client as __WebDAVclient


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

        call_args = chain(
            # positional args
            (f"{arg!r}" for arg in args[slice]),
            # keyword args
            (f"{k}:{v!r}" for k, v in kwargs.items()),
        )

        return f"{name}({', '.join(call_args)})"

    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