from json import JSONDecodeError 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, _, *args, **kwargs): """Return a cache key for use with cached methods.""" key = hashkey(name, *args, **kwargs) return hashkey(*(str(key_item) for key_item in key)) 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 , 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