64 lines
1.6 KiB
Python
64 lines
1.6 KiB
Python
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 <bytes>, 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,
|
|
)
|