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