Refactored simple cache and redis

Author: eadwinCode

ellar/cache/backends/aio_cache.py

@@ -83,7 +83,7 @@ def _cache_client(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=key.encode("utf-8"))
+        value = await self._cache_client.get(key.encode("utf-8"))
         if value:
             return self._deserializer(value)
         return None  # pragma: no cover
@@ -104,12 +104,12 @@ async def set_async(
 
     @make_key_decorator
     async def delete_async(self, key: str, version: str = None) -> bool:
-        return await self._cache_client.delete(key=key.encode("utf-8"))
+        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=key.encode("utf-8"), exptime=self.get_backend_timeout(timeout)
+            key.encode("utf-8"), exptime=self.get_backend_timeout(timeout)
         )

ellar/cache/backends/base.py

@@ -86,7 +86,7 @@ async def executor(self, func: t.Callable, *args: t.Any, **kwargs: t.Any) -> t.A
         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)
+        return await self.executor(self.get, key, version=version)
 
     async def set_async(
         self,
@@ -95,17 +95,19 @@ async def set_async(
         timeout: t.Union[float, int] = None,
         version: str = None,
     ) -> bool:
-        result = await self.executor(self.set, key, value, timeout)
+        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)
+        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)
+        result = await self.executor(self.touch, key, timeout=timeout, version=version)
         return bool(result)
 
     async def close_async(self, **kwargs: t.Any) -> None:

ellar/cache/backends/simple_cache.py renamed to ellar/cache/backends/local_cache.py

@@ -13,7 +13,7 @@
 from ..model import BaseCacheBackend
 
 
-class SimpleCacheBackendSync(IBaseCacheBackendAsync, ABC):
+class LocalMemCacheBackendSync(IBaseCacheBackendAsync, ABC):
     def _async_executor(self, func: t.Awaitable) -> t.Any:
         return get_or_create_eventloop().run_until_complete(func)
 
@@ -45,7 +45,7 @@ def touch(
         return bool(res)
 
 
-class SimpleCacheBackend(SimpleCacheBackendSync, BaseCacheBackend):
+class LocalMemCacheBackend(LocalMemCacheBackendSync, BaseCacheBackend):
     pickle_protocol = pickle.HIGHEST_PROTOCOL
 
     def __init__(self, **kwargs: t.Any) -> None:

ellar/cache/backends/redis/__init__.py

@@ -0,0 +1,4 @@
+from .backend import RedisCacheBackend
+from .serializer import IRedisSerializer
+
+__all__ = ["IRedisSerializer", "RedisCacheBackend"]

ellar/cache/backends/redis_cache.py renamed to ellar/cache/backends/redis/backend.py

@@ -1,23 +1,25 @@
-import asyncio
-import pickle
+import random
 import typing as t
 from abc import ABC
 
+from ellar.helper.event_loop import get_or_create_eventloop
+
 try:
     from redis.asyncio import Redis  # type: ignore
     from redis.asyncio.connection import ConnectionPool  # type: ignore
 except ImportError as e:  # pragma: no cover
     raise RuntimeError(
         "To use `RedisCacheBackend`, you have to install 'redis' package e.g. `pip install redis`"
     ) from e
-from ..interface import IBaseCacheBackendAsync
-from ..make_key_decorator import make_key_decorator
-from ..model import BaseCacheBackend
+from ...interface import IBaseCacheBackendAsync
+from ...make_key_decorator import make_key_decorator
+from ...model import BaseCacheBackend
+from .serializer import IRedisSerializer, RedisSerializer
 
 
 class RedisCacheBackendSync(IBaseCacheBackendAsync, ABC):
     def _async_executor(self, func: t.Awaitable) -> t.Any:
-        return asyncio.get_event_loop().run_until_complete(func)
+        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))
@@ -48,36 +50,58 @@ def touch(
 
 
 class RedisCacheBackend(RedisCacheBackendSync, BaseCacheBackend):
-    """Redis-based cache backend."""
+    """Redis-based cache backend.
+
+    Redis Server Construct example::
+        backend = RedisCacheBackend(servers=['redis://[[username]:[password]]@localhost:6379/0'])
+        OR
+        backend = RedisCacheBackend(servers=['redis://[[username]:[password]]@localhost:6379/0'])
+        OR
+        backend = RedisCacheBackend(servers=['rediss://[[username]:[password]]@localhost:6379/0'])
+        OR
+        backend = RedisCacheBackend(servers=['unix://[username@]/path/to/socket.sock?db=0[&password=password]'])
 
-    pickle_protocol: t.Any = pickle.HIGHEST_PROTOCOL
+    """
 
     def __init__(
         self,
-        url: str = "localhost",
-        db: int = None,
-        port: int = None,
-        username: str = None,
-        password: str = None,
+        servers: t.List[str],
         options: t.Dict = None,
-        serializer: t.Callable = pickle.dumps,
-        deserializer: t.Callable = pickle.loads,
+        serializer: IRedisSerializer = None,
         **kwargs: t.Any
     ) -> None:
         super().__init__(**kwargs)
 
-        self._cache_client_init: Redis = None
+        self._pools: t.Dict[int, ConnectionPool] = {}
+        self._servers = servers
         _default_options = options or {}
         self._options = {
-            "url": url,
-            "db": db,
-            "port": port,
-            "username": username,
-            "password": password,
             **_default_options,
         }
-        self._serializer = serializer
-        self._deserializer = deserializer
+        self._serializer = serializer or RedisSerializer()
+
+    def _get_connection_pool_index(self, write: bool) -> int:
+        # Write to the first server. Read from other servers if there are more,
+        # otherwise read from the first server.
+        if write or len(self._servers) == 1:
+            return 0
+        return random.randint(1, len(self._servers) - 1)
+
+    def _get_connection_pool(self, write: bool) -> ConnectionPool:
+        index = self._get_connection_pool_index(write)
+        if index not in self._pools:
+            self._pools[index] = ConnectionPool.from_url(
+                self._servers[index],
+                **self._options,
+            )
+        return self._pools[index]
+
+    def _get_client(self, *, write: bool = False) -> Redis:
+        # key is used so that the method signature remains the same and custom
+        # cache client can be implemented which might require the key to select
+        # the server, e.g. sharding.
+        pool = self._get_connection_pool(write)
+        return Redis(connection_pool=pool)
 
     def get_backend_timeout(
         self, timeout: t.Union[float, int] = None
@@ -88,21 +112,12 @@ def get_backend_timeout(
         # Non-positive values will cause the key to be deleted.
         return None if timeout is None else max(0, int(timeout))
 
-    @property
-    def _cache_client(self) -> Redis:
-        """
-        Implement transparent thread-safe access to a memcached client.
-        """
-        if self._cache_client_init is None:
-            pool = ConnectionPool.from_url(**self._options)
-            self._redis_int = Redis(connection_pool=pool)
-        return self._cache_client_init
-
     @make_key_decorator
     async def get_async(self, key: str, version: str = None) -> t.Any:
-        value = await self._cache_client.get(key)
+        client = self._get_client()
+        value = await client.get(key)
         if value:
-            return self._deserializer(value)
+            return self._serializer.load(value)
         return None
 
     @make_key_decorator
@@ -113,27 +128,26 @@ async def set_async(
         timeout: t.Union[float, int] = None,
         version: str = None,
     ) -> bool:
-        value = self._serializer(value, self.pickle_protocol)
+        client = self._get_client()
+        value = self._serializer.dumps(value)
         if timeout == 0:
-            await self._cache_client.delete(key)
+            await client.delete(key)
 
-        return bool(
-            await self._cache_client.set(
-                key, value, ex=self.get_backend_timeout(timeout)
-            )
-        )
+        return bool(await client.set(key, value, ex=self.get_backend_timeout(timeout)))
 
     @make_key_decorator
     async def delete_async(self, key: str, version: str = None) -> bool:
-        result = await self._cache_client.delete(key)
+        client = self._get_client()
+        result = await client.delete(key)
         return bool(result)
 
     @make_key_decorator
     async def touch_async(
         self, key: str, timeout: t.Union[float, int] = None, version: str = None
     ) -> bool:
+        client = self._get_client()
         if timeout is None:
-            res = await self._cache_client.persist(key)
+            res = await client.persist(key)
             return bool(res)
-        res = await self._cache_client.expire(key, self.get_backend_timeout(timeout))
+        res = await client.expire(key, self.get_backend_timeout(timeout))
         return bool(res)

ellar/cache/backends/redis/serializer.py

@@ -0,0 +1,33 @@
+import pickle
+import typing as t
+from abc import ABC, abstractmethod
+
+
+class IRedisSerializer(ABC):
+    default_protocol: int = pickle.HIGHEST_PROTOCOL
+
+    @abstractmethod
+    def load(self, data: t.Any) -> t.Any:
+        ...
+
+    @abstractmethod
+    def dumps(self, data: t.Any) -> t.Any:
+        ...
+
+
+class RedisSerializer(IRedisSerializer):
+    def __init__(self, protocol: int = None) -> None:
+        self._protocol = protocol or self.default_protocol
+
+    def load(self, data: t.Any) -> t.Any:
+        try:
+            return int(data)
+        except ValueError:
+            return pickle.loads(data)
+
+    def dumps(self, data: t.Any) -> t.Any:
+        # Only skip pickling for integers, a int subclasses as bool should be
+        # pickled.
+        if type(data) is int:
+            return data
+        return pickle.dumps(data, self._protocol)

ellar/cache/model.py

@@ -27,13 +27,13 @@ async def has_key_async(self, key: str, version: str = None) -> bool:
         """
         Return True if the key is in the cache and has not expired.
         """
-        return await self.get_async(key, version) is not None
+        return await self.get_async(key, version=version) is not None
 
     def has_key(self, key: str, version: str = None) -> bool:
         """
         Return True if the key is in the cache and has not expired.
         """
-        return self.get(key, version) is not None
+        return self.get(key, version=version) is not None
 
     def validate_key(self, key: str) -> None:
         if len(key) > self.MEMCACHE_MAX_KEY_LENGTH:

ellar/cache/service.py

@@ -2,7 +2,7 @@
 
 from ellar.di import injectable
 
-from .backends.simple_cache import SimpleCacheBackend
+from .backends.local_cache import LocalMemCacheBackend
 from .interface import ICacheService, ICacheServiceSync
 from .model import BaseCacheBackend
 
@@ -52,7 +52,7 @@ def __init__(self, backends: t.Dict[str, BaseCacheBackend] = None) -> None:
                 "default"
             ), "CACHES configuration must have a 'default' key."
         self._backends = backends or {
-            "default": SimpleCacheBackend(key_prefix="ellar", version=1, timeout=300)
+            "default": LocalMemCacheBackend(key_prefix="ellar", version=1, timeout=300)
         }
 
     def _get_backend(self, backend: str = None) -> BaseCacheBackend: