54 lines
1.4 KiB
Python
54 lines
1.4 KiB
Python
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 <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
|