Merge pull request #65 from eadwinCode/cache_feat

Cache

Author: eadwinCode

.flake8

@@ -1,6 +1,6 @@
 [flake8]
 max-line-length = 88
-ignore = E203, E241, E501, W503, F811
+ignore = E203, E241, E501, W503, F811, W601
 exclude =
     .git,
     __pycache__

docs/caching.md

@@ -0,0 +1,5 @@
+from .interface import ICacheService
+from .model import BaseCacheBackend
+from .service import CacheService
+
+__all__ = ["CacheService", "ICacheService", "BaseCacheBackend"]

ellar/cache/__init__.py

@@ -0,0 +1,116 @@
+import pickle
+import typing as t
+from abc import ABC
+
+from ellar.helper.event_loop import get_or_create_eventloop
+
+try:
+    from aiomcache import Client
+except ImportError as e:  # pragma: no cover
+    raise RuntimeError(
+        "To use `AioMemCacheBackend`, you have to install 'aiomcache' package e.g. `pip install aiomcache`"
+    ) from e
+
+
+from ..interface import IBaseCacheBackendAsync
+from ..make_key_decorator import make_key_decorator, make_key_decorator_and_validate
+from ..model import BaseCacheBackend
+
+
+class _AioMemCacheBackendSync(IBaseCacheBackendAsync, ABC):
+    def _async_executor(self, func: t.Awaitable) -> t.Any:
+        return get_or_create_eventloop().run_until_complete(func)
+
+    def get(self, key: str, version: str = None) -> t.Any:
+        return self._async_executor(self.get_async(key, version=version))
+
+    def delete(self, key: str, version: str = None) -> bool:
+        res = self._async_executor(self.delete_async(key, version=version))
+        return bool(res)
+
+    def set(
+        self,
+        key: str,
+        value: t.Any,
+        timeout: t.Union[float, int] = None,
+        version: str = None,
+    ) -> bool:
+        res = self._async_executor(
+            self.set_async(key, value, version=version, timeout=timeout)
+        )
+        return bool(res)
+
+    def touch(
+        self, key: str, timeout: t.Union[float, int] = None, version: str = None
+    ) -> bool:
+        res = self._async_executor(
+            self.touch_async(key, version=version, timeout=timeout)
+        )
+        return bool(res)
+
+
+class AioMemCacheBackend(_AioMemCacheBackendSync, BaseCacheBackend):
+    """Memcached-based cache backend."""
+
+    pickle_protocol = pickle.HIGHEST_PROTOCOL
+    MEMCACHE_CLIENT: t.Type = Client
+
+    def __init__(
+        self,
+        host: str,
+        port: int = 11211,
+        pool_size: int = 2,
+        pool_minsize: int = None,
+        serializer: t.Callable = pickle.dumps,
+        deserializer: t.Callable = pickle.loads,
+        **kwargs: t.Any
+    ) -> None:
+        super().__init__(**kwargs)
+        self._client: Client = None  # type: ignore[assignment]
+        self._client_options = dict(
+            host=host, port=port, pool_size=pool_size, pool_minsize=pool_minsize
+        )
+        self._serializer = serializer
+        self._deserializer = deserializer
+
+    def get_backend_timeout(self, timeout: t.Union[float, int] = None) -> int:
+        return int(super().get_backend_timeout(timeout))
+
+    @property
+    def _cache_client(self) -> Client:
+        if self._client is None:
+            self._client = self.MEMCACHE_CLIENT(**self._client_options)
+        return self._client
+
+    @make_key_decorator
+    async def get_async(self, key: str, version: str = None) -> t.Optional[t.Any]:
+        value = await self._cache_client.get(key.encode("utf-8"))
+        if value:
+            return self._deserializer(value)
+        return None  # pragma: no cover
+
+    @make_key_decorator_and_validate
+    async def set_async(
+        self,
+        key: str,
+        value: t.Any,
+        timeout: t.Union[float, int] = None,
+        version: str = None,
+    ) -> bool:
+        return await self._cache_client.set(
+            key.encode("utf-8"),
+            self._serializer(value, self.pickle_protocol),
+            exptime=self.get_backend_timeout(timeout),
+        )
+
+    @make_key_decorator
+    async def delete_async(self, key: str, version: str = None) -> bool:
+        return await self._cache_client.delete(key.encode("utf-8"))
+
+    @make_key_decorator
+    async def touch_async(
+        self, key: str, timeout: t.Union[float, int] = None, version: str = None
+    ) -> bool:
+        return await self._cache_client.touch(
+            key.encode("utf-8"), exptime=self.get_backend_timeout(timeout)
+        )

ellar/cache/backends/__init__.py

@@ -0,0 +1,113 @@
+import typing as t
+from abc import ABC
+
+from starlette.concurrency import run_in_threadpool
+
+from ..make_key_decorator import make_key_decorator, make_key_decorator_and_validate
+from ..model import BaseCacheBackend
+
+
+class _BasePylibMemcachedCacheSync(BaseCacheBackend, ABC):
+    _cache_client: t.Any
+
+    @make_key_decorator
+    def get(self, key: str, version: str = None) -> t.Any:
+        return self._cache_client.get(key)
+
+    @make_key_decorator_and_validate
+    def set(
+        self,
+        key: str,
+        value: t.Any,
+        timeout: t.Union[float, int] = None,
+        version: str = None,
+    ) -> bool:
+        result = self._cache_client.set(
+            key, value, int(self.get_backend_timeout(timeout))
+        )
+        if not result:
+            # Make sure the key doesn't keep its old value in case of failure
+            # to set (memcached's 1MB limit).
+            self._cache_client.delete(key)
+            return False
+        return bool(result)
+
+    @make_key_decorator
+    def delete(self, key: str, version: str = None) -> bool:
+        result = self._cache_client.delete(key)
+        return bool(result)
+
+    @make_key_decorator
+    def touch(
+        self, key: str, timeout: t.Union[float, int] = None, version: str = None
+    ) -> bool:
+        result = self._cache_client.touch(key, self.get_backend_timeout(timeout))
+        return bool(result)
+
+    def close(self, **kwargs: t.Any) -> None:
+        # Many clients don't clean up connections properly.
+        self._cache_client.disconnect_all()
+
+    def clear(self) -> None:
+        self._cache_client.flush_all()
+
+
+class BasePylibMemcachedCache(_BasePylibMemcachedCacheSync):
+    MEMCACHE_CLIENT: t.Type
+
+    def __init__(self, servers: t.List[str], options: t.Dict = None, **kwargs: t.Any):
+        super().__init__(**kwargs)
+        self._servers = servers
+
+        self._cache_client_init: t.Any = None
+        self._options = options or {}
+
+    @property
+    def _cache_client(self) -> t.Any:
+        """
+        Implement transparent thread-safe access to a memcached client.
+        """
+        if self._cache_client_init is None:
+            self._cache_client_init = self.MEMCACHE_CLIENT(
+                self._servers, **self._options
+            )
+        return self._cache_client_init
+
+    async def executor(self, func: t.Callable, *args: t.Any, **kwargs: t.Any) -> t.Any:
+        return await run_in_threadpool(func, *args, **kwargs)
+
+    async def get_async(self, key: str, version: str = None) -> t.Any:
+        return await self.executor(self.get, key, version=version)
+
+    async def set_async(
+        self,
+        key: str,
+        value: t.Any,
+        timeout: t.Union[float, int] = None,
+        version: str = None,
+    ) -> bool:
+        result = await self.executor(
+            self.set, key, value, timeout=timeout, version=version
+        )
+        return bool(result)
+
+    async def delete_async(self, key: str, version: str = None) -> bool:
+        result = await self.executor(self.delete, key, version=version)
+        return bool(result)
+
+    async def touch_async(
+        self, key: str, timeout: t.Union[float, int] = None, version: str = None
+    ) -> bool:
+        result = await self.executor(self.touch, key, timeout=timeout, version=version)
+        return bool(result)
+
+    async def close_async(self, **kwargs: t.Any) -> None:
+        # Many clients don't clean up connections properly.
+        await self.executor(self._cache_client.disconnect_all)
+
+    async def clear_async(self) -> None:
+        await self.executor(self._cache_client.flush_all)
+
+    def validate_key(self, key: str) -> None:
+        super().validate_key(key)
+        self._memcache_key_warnings(key)

ellar/cache/backends/aio_cache.py

@@ -0,0 +1,118 @@
+import pickle
+import time
+import typing as t
+from abc import ABC
+from collections import OrderedDict
+
+from anyio import Lock
+
+from ellar.helper.event_loop import get_or_create_eventloop
+
+from ..interface import IBaseCacheBackendAsync
+from ..make_key_decorator import make_key_decorator, make_key_decorator_and_validate
+from ..model import BaseCacheBackend
+
+
+class _LocalMemCacheBackendSync(IBaseCacheBackendAsync, ABC):
+    def _async_executor(self, func: t.Awaitable) -> t.Any:
+        return get_or_create_eventloop().run_until_complete(func)
+
+    def get(self, key: str, version: str = None) -> t.Any:
+        return self._async_executor(self.get_async(key, version=version))
+
+    def delete(self, key: str, version: str = None) -> bool:
+        res = self._async_executor(self.delete_async(key, version=version))
+        return bool(res)
+
+    def set(
+        self,
+        key: str,
+        value: t.Any,
+        timeout: t.Union[float, int] = None,
+        version: str = None,
+    ) -> bool:
+        res = self._async_executor(
+            self.set_async(key, value, timeout=timeout, version=version)
+        )
+        return bool(res)
+
+    def touch(
+        self, key: str, timeout: t.Union[float, int] = None, version: str = None
+    ) -> bool:
+        res = self._async_executor(
+            self.touch_async(key, timeout=timeout, version=version)
+        )
+        return bool(res)
+
+
+class LocalMemCacheBackend(_LocalMemCacheBackendSync, BaseCacheBackend):
+    pickle_protocol = pickle.HIGHEST_PROTOCOL
+
+    def __init__(self, **kwargs: t.Any) -> None:
+        super().__init__(**kwargs)
+        self._cache: t.Dict[str, bytes] = OrderedDict()
+        self._expire_track: t.Dict[str, float] = {}
+        self._lock = Lock()
+
+    @make_key_decorator
+    async def get_async(self, key: str, version: str = None) -> t.Any:
+        async with self._lock:
+            if self._has_expired(key):
+                await self._delete(key)
+                return None
+
+            pickled = self._cache[key]
+            return pickle.loads(pickled)
+
+    async def _delete(self, key: str) -> bool:
+        try:
+            self._cache.pop(key)
+            self._expire_track.pop(key)
+        except KeyError:
+            return False
+        return True
+
+    @make_key_decorator
+    async def delete_async(self, key: str, version: str = None) -> bool:
+        async with self._lock:
+            return await self._delete(key)
+
+    @make_key_decorator_and_validate
+    async def set_async(
+        self,
+        key: str,
+        value: t.Any,
+        timeout: t.Union[float, int] = None,
+        version: str = None,
+    ) -> bool:
+        async with self._lock:
+            self._cache[key] = pickle.dumps(value, self.pickle_protocol)
+            self._expire_track[key] = self.get_backend_timeout(timeout)
+            return True
+
+    def _has_expired(self, key: str) -> bool:
+        exp = self._expire_track.get(key, -1)
+        return exp is not None and exp <= time.time()
+
+    @make_key_decorator
+    async def has_key_async(self, key: str, version: str = None) -> bool:
+        async with self._lock:
+            if self._has_expired(key):
+                await self._delete(key)
+                return False
+            return True
+
+    @make_key_decorator
+    async def touch_async(
+        self, key: str, timeout: t.Union[float, int] = None, version: str = None
+    ) -> bool:
+        async with self._lock:
+            if self._has_expired(key):
+                return False
+
+            self._expire_track[key] = self.get_backend_timeout(timeout)
+            return True
+
+    def has_key(self, key: str, version: str = None) -> bool:
+        res = self._async_executor(self.has_key_async(key, version=version))
+        return bool(res)

ellar/cache/backends/base.py

@@ -0,0 +1,30 @@
+"""
+PyLibMCCacheBackend inspired by Django PyLibMCCache
+"""
+import typing as t
+
+try:
+    from pylibmc import Client
+except ImportError as e:  # pragma: no cover
+    raise RuntimeError(
+        "To use `PyLibMCCacheBackend`, you have to install 'pylibmc' package e.g. `pip install pylibmc`"
+    ) from e
+
+from .base import BasePylibMemcachedCache
+
+
+class PyLibMCCacheBackend(BasePylibMemcachedCache):
+    """An implementation of a cache binding using pylibmc"""
+
+    MEMCACHE_CLIENT = Client
+
+    def __init__(self, servers: t.List[str], options: t.Dict = None, **kwargs: t.Any):
+        super().__init__(servers, options=options, **kwargs)
+
+    async def close_async(self, **kwargs: t.Any) -> None:
+        # libmemcached manages its own connections. Don't call disconnect_all()
+        # as it resets the failover state and creates unnecessary reconnects.
+        return None
+
+    def close(self, **kwargs: t.Any) -> None:
+        return None