1  import typing as _t
   2  from time import time
   3  
   4  from cachelib.base import BaseCache
   5  from cachelib.serializers import SimpleSerializer
   6  
   7  
   8  class SimpleCache(BaseCache):
   9      """Simple memory cache for single process environments.  This class exists
  10      mainly for the development server and is not 100% thread safe.  It tries
  11      to use as many atomic operations as possible and no locks for simplicity
  12      but it could happen under heavy load that keys are added multiple times.
  13  
  14      :param threshold: the maximum number of items the cache stores before
  15                        it starts deleting some.
  16      :param default_timeout: the default timeout that is used if no timeout is
  17                              specified on :meth:`~BaseCache.set`. A timeout of
  18                              0 indicates that the cache never expires.
  19      """
  20  
  21      serializer = SimpleSerializer()
  22  
  23      def __init__(
  24          self,
  25          threshold: int = 500,
  26          default_timeout: int = 300,
  27      ):
  28          BaseCache.__init__(self, default_timeout)
  29          self._cache: _t.Dict[str, _t.Any] = {}
  30          self._threshold = threshold or 500  # threshold = 0
  31  
  32      def _over_threshold(self) -> bool:
  33          return len(self._cache) > self._threshold
  34  
  35      def _remove_expired(self, now: float) -> None:
  36          toremove = [k for k, (expires, _) in self._cache.items() if expires < now]
  37          for k in toremove:
  38              self._cache.pop(k, None)
  39  
  40      def _remove_older(self) -> None:
  41          k_ordered = (
  42              k for k, v in sorted(self._cache.items(), key=lambda item: item[1][0])
  43          )
  44          for k in k_ordered:
  45              self._cache.pop(k, None)
  46              if not self._over_threshold():
  47                  break
  48  
  49      def _prune(self) -> None:
  50          if self._over_threshold():
  51              now = time()
  52              self._remove_expired(now)
  53          # remove older items if still over threshold
  54          if self._over_threshold():
  55              self._remove_older()
  56  
  57      def _normalize_timeout(self, timeout: _t.Optional[int]) -> int:
  58          timeout = BaseCache._normalize_timeout(self, timeout)
  59          if timeout > 0:
  60              timeout = int(time()) + timeout
  61          return timeout
  62  
  63      def get(self, key: str) -> _t.Any:
  64          try:
  65              expires, value = self._cache[key]
  66              if expires == 0 or expires > time():
  67                  return self.serializer.loads(value)
  68          except KeyError:
  69              return None
  70  
  71      def set(
  72          self, key: str, value: _t.Any, timeout: _t.Optional[int] = None
  73      ) -> _t.Optional[bool]:
  74          expires = self._normalize_timeout(timeout)
  75          self._prune()
  76          self._cache[key] = (expires, self.serializer.dumps(value))
  77          return True
  78  
  79      def add(self, key: str, value: _t.Any, timeout: _t.Optional[int] = None) -> bool:
  80          expires = self._normalize_timeout(timeout)
  81          self._prune()
  82          item = (expires, self.serializer.dumps(value))
  83          if key in self._cache:
  84              return False
  85          self._cache.setdefault(key, item)
  86          return True
  87  
  88      def delete(self, key: str) -> bool:
  89          return self._cache.pop(key, None) is not None
  90  
  91      def has(self, key: str) -> bool:
  92          try:
  93              expires, value = self._cache[key]
  94              return bool(expires == 0 or expires > time())
  95          except KeyError:
  96              return False
  97  
  98      def clear(self) -> bool:
  99          self._cache.clear()
 100          return not bool(self._cache)
 101