1  import platform
   2  import typing as _t
   3  
   4  from cachelib.base import BaseCache
   5  from cachelib.serializers import UWSGISerializer
   6  
   7  
   8  class UWSGICache(BaseCache):
   9      """Implements the cache using uWSGI's caching framework.
  10  
  11      .. note::
  12          This class cannot be used when running under PyPy, because the uWSGI
  13          API implementation for PyPy is lacking the needed functionality.
  14  
  15      :param default_timeout: The default timeout in seconds.
  16      :param cache: The name of the caching instance to connect to, for
  17          example: mycache@localhost:3031, defaults to an empty string, which
  18          means uWSGI will cache in the local instance. If the cache is in the
  19          same instance as the werkzeug app, you only have to provide the name of
  20          the cache.
  21      """
  22  
  23      serializer = UWSGISerializer()
  24  
  25      def __init__(
  26          self,
  27          default_timeout: int = 300,
  28          cache: str = "",
  29      ):
  30          BaseCache.__init__(self, default_timeout)
  31  
  32          if platform.python_implementation() == "PyPy":
  33              raise RuntimeError(
  34                  "uWSGI caching does not work under PyPy, see "
  35                  "the docs for more details."
  36              )
  37  
  38          try:
  39              import uwsgi  # type: ignore
  40  
  41              self._uwsgi = uwsgi
  42          except ImportError as err:
  43              raise RuntimeError(
  44                  "uWSGI could not be imported, are you running under uWSGI?"
  45              ) from err
  46  
  47          self.cache = cache
  48  
  49      def get(self, key: str) -> _t.Any:
  50          rv = self._uwsgi.cache_get(key, self.cache)
  51          if rv is None:
  52              return
  53          return self.serializer.loads(rv)
  54  
  55      def delete(self, key: str) -> bool:
  56          return bool(self._uwsgi.cache_del(key, self.cache))
  57  
  58      def set(
  59          self, key: str, value: _t.Any, timeout: _t.Optional[int] = None
  60      ) -> _t.Optional[bool]:
  61          result = self._uwsgi.cache_update(
  62              key,
  63              self.serializer.dumps(value),
  64              self._normalize_timeout(timeout),
  65              self.cache,
  66          )  # type: bool
  67          return result
  68  
  69      def add(self, key: str, value: _t.Any, timeout: _t.Optional[int] = None) -> bool:
  70          return bool(
  71              self._uwsgi.cache_set(
  72                  key,
  73                  self.serializer.dumps(value),
  74                  self._normalize_timeout(timeout),
  75                  self.cache,
  76              )
  77          )
  78  
  79      def clear(self) -> bool:
  80          return bool(self._uwsgi.cache_clear(self.cache))
  81  
  82      def has(self, key: str) -> bool:
  83          return self._uwsgi.cache_exists(key, self.cache) is not None
  84