Lenaisten/advent22
1
0
Fork 0
mirror of https://code.lenaisten.de/Lenaisten/advent22.git synced 2024-11-23 00:03:07 +00:00
Code Issues Projects Releases Packages Wiki Activity

simpler helpers.davkey

Browse source
This commit is contained in:
Jörn-Michael Miehe 2023-11-09 12:55:10 +01:00
parent eef38502c6
commit 02ba2c67a5
2 changed files with 15 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
api/advent22_api/core/dav/helpers.py
View file

@ -1,4 +1,5 @@
from json import JSONDecodeError from json import JSONDecodeError
from typing import Callable, Hashable
import requests import requests
from cachetools.keys import hashkey from cachetools.keys import hashkey
@ -8,12 +9,18 @@ from redis.typing import EncodableT
from webdav3.client import Client as __WebDAVclient from webdav3.client import Client as __WebDAVclient
def davkey(name, _, *args, **kwargs): 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.""" """Return a cache key for use with cached methods."""
key = hashkey(name, *args, **kwargs) key = hashkey(name, *args[slice], **kwargs)
return hashkey(*(str(key_item) for key_item in key)) return hashkey(*(str(key_item) for key_item in key))
return func
class WebDAVclient(__WebDAVclient): class WebDAVclient(__WebDAVclient):
def execute_request( def execute_request(

18
api/advent22_api/core/dav/webdav.py
View file

@ -1,4 +1,3 @@
import functools
import logging import logging
import re import re
from io import BytesIO from io import BytesIO
@ -34,10 +33,7 @@ class WebDAV:
@classmethod @classmethod
@asyncify @asyncify
@cachedmethod( @cachedmethod(cache=lambda cls: cls._cache, key=davkey("list_files"))
cache=lambda cls: cls._cache,
key=functools.partial(davkey, "list_files"),
)
def list_files( def list_files(
cls, cls,
directory: str = "", directory: str = "",
@ -55,10 +51,7 @@ class WebDAV:
@classmethod @classmethod
@asyncify @asyncify
@cachedmethod( @cachedmethod(cache=lambda cls: cls._cache, key=davkey("exists"))
cache=lambda cls: cls._cache,
key=functools.partial(davkey, "exists"),
)
def exists(cls, path: str) -> bool: def exists(cls, path: str) -> bool:
""" """
`True` iff there is a WebDAV resource at `path` `True` iff there is a WebDAV resource at `path`
@ -69,10 +62,7 @@ class WebDAV:
@classmethod @classmethod
@asyncify @asyncify
@cachedmethod( @cachedmethod(cache=lambda cls: cls._cache, key=davkey("read_bytes"))
cache=lambda cls: cls._cache,
key=functools.partial(davkey, "read_bytes"),
)
def read_bytes(cls, path: str) -> bytes: def read_bytes(cls, path: str) -> bytes:
""" """
Load WebDAV file from `path` as bytes Load WebDAV file from `path` as bytes
@ -105,7 +95,7 @@ class WebDAV:
cls._webdav_client.upload_to(buffer, path) cls._webdav_client.upload_to(buffer, path)
# invalidate cache entry # invalidate cache entry
cls._cache.pop(davkey("read_bytes", None, path)) cls._cache.pop(davkey("read_bytes")(path))
@classmethod @classmethod
async def write_str(cls, path: str, content: str, encoding="utf-8") -> None: async def write_str(cls, path: str, content: str, encoding="utf-8") -> None: