Skip to content

Commit

Permalink
Improve adapters/django typings
Browse files Browse the repository at this point in the history
  • Loading branch information
k0t3n committed Jul 24, 2024
1 parent 55f5a71 commit 10cd779
Showing 1 changed file with 20 additions and 19 deletions.
39 changes: 20 additions & 19 deletions theine/adapters/django.py
Original file line number Diff line number Diff line change
@@ -1,45 +1,46 @@
from datetime import timedelta
from threading import Lock
from typing import Optional, cast
from typing import Any, Callable, Optional, Union, cast, Mapping

from django.core.cache.backends.base import DEFAULT_TIMEOUT, BaseCache

from theine import Cache as Theine
from theine.theine import sentinel

KEY_TYPE = Union[str, Callable[..., str]]
VALUE_TYPE = Any
VERSION_TYPE = Optional[int]


class Cache(BaseCache):
def __init__(self, name, params):
def __init__(self, name: str, params: dict[str, Any]):
super().__init__(params)
options = params.get("OPTIONS", {})
policy = options.get("POLICY", "tlfu")
self.cache = Theine(policy, self._max_entries)

def _timeout_seconds(self, timeout) -> Optional[float]:
if timeout == DEFAULT_TIMEOUT:
return self.default_timeout
return timeout

def add(self, key, value, timeout=DEFAULT_TIMEOUT, version=None):
def add(self, key: KEY_TYPE, value: VALUE_TYPE, timeout: Optional[float] = DEFAULT_TIMEOUT,
version: VERSION_TYPE = None) -> bool:
data = self.get(key, sentinel, version)
if data is not sentinel:
return False
key = self.make_key(key, version)
backend_timeout = self.get_backend_timeout(timeout)
self.cache.set(
key,
value,
timedelta(seconds=cast(float, timeout))
if timeout is not DEFAULT_TIMEOUT
else None,
timedelta(seconds=backend_timeout) if backend_timeout else None,
)
return True

def get(self, key, default=None, version=None):
def get(self, key: KEY_TYPE, default: Optional[VALUE_TYPE] = None, version: VERSION_TYPE = None) -> Optional[
VALUE_TYPE]:
key = self.make_key(key, version)
return self.cache.get(key, default)

def set(self, key, value, timeout=DEFAULT_TIMEOUT, version=None):
to = self._timeout_seconds(timeout)
def set(self, key: KEY_TYPE, value: VALUE_TYPE, timeout: Optional[float] = DEFAULT_TIMEOUT,
version: VERSION_TYPE = None) -> None:
to = self.get_backend_timeout(timeout)
if to is not None and to <= 0:
self.delete(key)
return
Expand All @@ -50,26 +51,26 @@ def set(self, key, value, timeout=DEFAULT_TIMEOUT, version=None):
timedelta(seconds=to) if to is not None else None,
)

def touch(self, key, timeout=DEFAULT_TIMEOUT, version=None):
def touch(self, key: KEY_TYPE, timeout: Optional[float] = DEFAULT_TIMEOUT, version: VERSION_TYPE = None) -> bool:
data = self.get(key, sentinel, version)
if data is sentinel:
return False
nkey = self.make_key(key, version)
if (
timeout is not DEFAULT_TIMEOUT
and timeout is not None
and cast(float, timeout) <= 0
and timeout <= 0
):
self.cache.delete(nkey)
return True
to = self._timeout_seconds(timeout)
to = self.get_backend_timeout(timeout)
with Lock():
self.cache._access(nkey, timedelta(seconds=to) if to is not None else None)
return True

def delete(self, key, version=None):
def delete(self, key: KEY_TYPE, version: VERSION_TYPE = None) -> bool:
key = self.make_key(key, version)
return self.cache.delete(key)

def clear(self):
def clear(self) -> None:
self.cache.clear()

0 comments on commit 10cd779

Please sign in to comment.