Fixed event loop bug with synchronous call

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__

ellar/cache/backends/aio_cache.py

@@ -1,8 +1,9 @@
-import asyncio
 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
@@ -18,7 +19,7 @@
 
 class AioMemCacheBackendSync(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))
@@ -28,14 +29,20 @@ def delete(self, key: str, version: str = None) -> bool:
         return bool(res)
 
     def set(
-        self, key: str, value: t.Any, timeout: int = None, version: str = None
+        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: int = None, version: str = None) -> bool:
+    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)
         )
@@ -65,7 +72,7 @@ def __init__(
         self._serializer = serializer
         self._deserializer = deserializer
 
-    def get_backend_timeout(self, timeout: int = None) -> int:
+    def get_backend_timeout(self, timeout: t.Union[float, int] = None) -> int:
         return int(super().get_backend_timeout(timeout))
 
     @property
@@ -83,7 +90,11 @@ async def get_async(self, key: str, version: str = None) -> t.Optional[t.Any]:
 
     @make_key_decorator
     async def set_async(
-        self, key: str, value: t.Any, timeout: int = None, version: str = None
+        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"),
@@ -97,7 +108,7 @@ async def delete_async(self, key: str, version: str = None) -> bool:
 
     @make_key_decorator
     async def touch_async(
-        self, key: str, timeout: int = None, version: str = None
+        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)

ellar/cache/backends/base.py

@@ -11,12 +11,16 @@ class BasePylibMemcachedCacheSync(BaseCacheBackend, ABC):
     _cache_client: t.Any
 
     @make_key_decorator
-    async def get(self, key: str, version: str = None) -> t.Any:
+    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: int = None, version: str = None
+        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))
@@ -34,7 +38,9 @@ def delete(self, key: str, version: str = None) -> bool:
         return bool(result)
 
     @make_key_decorator
-    def touch(self, key: str, timeout: int = None, version: str = None) -> bool:
+    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)
 
@@ -83,7 +89,11 @@ async def get_async(self, key: str, version: str = None) -> t.Any:
         return await self.executor(self.get, key)
 
     async def set_async(
-        self, key: str, value: t.Any, timeout: int = None, version: str = None
+        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)
         return bool(result)
@@ -93,7 +103,7 @@ async def delete_async(self, key: str, version: str = None) -> bool:
         return bool(result)
 
     async def touch_async(
-        self, key: str, timeout: int = None, version: str = None
+        self, key: str, timeout: t.Union[float, int] = None, version: str = None
     ) -> bool:
         result = await self.executor(self.touch, key, timeout)
         return bool(result)

ellar/cache/backends/redis_cache.py

@@ -27,14 +27,20 @@ def delete(self, key: str, version: str = None) -> bool:
         return bool(res)
 
     def set(
-        self, key: str, value: t.Any, timeout: int = None, version: str = None
+        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: int = None, version: str = None) -> bool:
+    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)
         )
@@ -73,7 +79,9 @@ def __init__(
         self._serializer = serializer
         self._deserializer = deserializer
 
-    def get_backend_timeout(self, timeout: int = None) -> t.Union[float, int]:
+    def get_backend_timeout(
+        self, timeout: t.Union[float, int] = None
+    ) -> t.Union[float, int]:
         if timeout is None:
             timeout = self._default_timeout
         # The key will be made persistent if None used as a timeout.
@@ -99,7 +107,11 @@ async def get_async(self, key: str, version: str = None) -> t.Any:
 
     @make_key_decorator
     async def set_async(
-        self, key: str, value: t.Any, timeout: int = None, version: str = None
+        self,
+        key: str,
+        value: t.Any,
+        timeout: t.Union[float, int] = None,
+        version: str = None,
     ) -> bool:
         value = self._serializer(value, self.pickle_protocol)
         if timeout == 0:
@@ -118,7 +130,7 @@ async def delete_async(self, key: str, version: str = None) -> bool:
 
     @make_key_decorator
     async def touch_async(
-        self, key: str, timeout: int = None, version: str = None
+        self, key: str, timeout: t.Union[float, int] = None, version: str = None
     ) -> bool:
         if timeout is None:
             res = await self._cache_client.persist(key)

ellar/cache/backends/simple_cache.py

@@ -1,4 +1,3 @@
-import asyncio
 import pickle
 import time
 import typing as t
@@ -7,14 +6,16 @@
 
 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 SimpleCacheBackendSync(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))
@@ -24,14 +25,20 @@ def delete(self, key: str, version: str = None) -> bool:
         return bool(res)
 
     def set(
-        self, key: str, value: t.Any, timeout: int = None, version: str = None
+        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: int = None, version: str = None) -> bool:
+    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)
         )
@@ -72,7 +79,11 @@ async def delete_async(self, key: str, version: str = None) -> bool:
 
     @make_key_decorator_and_validate
     async def set_async(
-        self, key: str, value: t.Any, timeout: int = None, version: str = None
+        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)
@@ -93,7 +104,7 @@ async def has_key_async(self, key: str, version: str = None) -> bool:
 
     @make_key_decorator
     async def touch_async(
-        self, key: str, timeout: int = None, version: str = None
+        self, key: str, timeout: t.Union[float, int] = None, version: str = None
     ) -> bool:
         async with self._lock:
             if self._has_expired(key):

ellar/cache/interface.py

@@ -22,7 +22,11 @@ async def delete_async(self, key: str, version: str = None) -> bool:
 
     @abstractmethod
     async def set_async(
-        self, key: str, value: t.Any, timeout: int = None, version: str = None
+        self,
+        key: str,
+        value: t.Any,
+        timeout: t.Union[float, int] = None,
+        version: str = None,
     ) -> bool:
         """Add a new key/value to the cache (overwrites value, if key already
         exists in the cache).
@@ -40,7 +44,7 @@ async def set_async(
 
     @abstractmethod
     async def touch_async(
-        self, key: str, timeout: int = None, version: str = None
+        self, key: str, timeout: t.Union[float, int] = None, version: str = None
     ) -> bool:
         """
         Update the key's expiry time using timeout. Return True if successful
@@ -68,7 +72,11 @@ def delete(self, key: str, version: str = None) -> bool:
 
     @abstractmethod
     def set(
-        self, key: str, value: t.Any, timeout: int = None, version: str = None
+        self,
+        key: str,
+        value: t.Any,
+        timeout: t.Union[float, int] = None,
+        version: str = None,
     ) -> bool:
         """Add a new key/value to the cache (overwrites value, if key already
         exists in the cache).
@@ -85,7 +93,9 @@ def set(
         """
 
     @abstractmethod
-    def touch(self, key: str, timeout: int = None, version: str = None) -> bool:
+    def touch(
+        self, key: str, timeout: t.Union[float, int] = None, version: str = None
+    ) -> bool:
         """
         Update the key's expiry time using timeout. Return True if successful
         or False if the key does not exist.
@@ -117,7 +127,7 @@ def set(
         self,
         key: str,
         value: t.Any,
-        timeout: int = None,
+        timeout: t.Union[float, int] = None,
         version: str = None,
         backend: str = None,
     ) -> bool:
@@ -138,7 +148,11 @@ def set(
 
     @abstractmethod
     def touch(
-        self, key: str, timeout: int = None, version: str = None, backend: str = None
+        self,
+        key: str,
+        timeout: t.Union[float, int] = None,
+        version: str = None,
+        backend: str = None,
     ) -> bool:
         """
         Update the key's expiry time using timeout. Return True if successful
@@ -181,7 +195,7 @@ async def set_async(
         self,
         key: str,
         value: t.Any,
-        timeout: int = None,
+        timeout: t.Union[float, int] = None,
         version: str = None,
         backend: str = None,
     ) -> bool:
@@ -202,7 +216,11 @@ async def set_async(
 
     @abstractmethod
     async def touch_async(
-        self, key: str, timeout: int = None, version: str = None, backend: str = None
+        self,
+        key: str,
+        timeout: t.Union[float, int] = None,
+        version: str = None,
+        backend: str = None,
     ) -> bool:
         """
         Update the key's expiry time using timeout. Return True if successful

ellar/cache/model.py

@@ -61,7 +61,9 @@ def make_key(self, key: str, version: str = None) -> str:
         """
         return "%s:%s:%s" % (self._key_prefix, version or self._version, key)
 
-    def get_backend_timeout(self, timeout: int = None) -> t.Union[float, int]:
+    def get_backend_timeout(
+        self, timeout: t.Union[float, int] = None
+    ) -> t.Union[float, int]:
         """
         Return the timeout value usable by this backend based upon the provided
         timeout.