Implement proper caching closing

retribute_api/cache.py

@@ -15,6 +15,9 @@ class Backend:
     async def set(self, key):
         raise NotImplementedError
 
+    async def close(self):
+        return
+
 
 class Dummy(Backend):
     def __init__(self):
@@ -59,6 +62,11 @@ class Redis(Backend):
         r = await self.redis()
         await r.set(key, json.dumps(value))
 
+    async def close(self):
+        if not self._redis:
+            return
+        await self._redis.close()
+
 
 def get_default():
     return Redis(settings.ASYNC_REDIS_PARAMS)

retribute_api/search/consumers.py

@@ -36,6 +36,18 @@ def wrapper_500(callback):
     return callback
 
 
+def with_cache(f):
+    async def inner(*args, **kwargs):
+        c = kwargs.setdefault("cache", cache.get_default())
+        try:
+            return await f(*args, **kwargs)
+        except Exception:
+            await c.close()
+            raise
+
+    return inner
+
+
 def ignore_aiohttp_ssl_eror(loop, aiohttpversion="3.5.4"):
     """Ignore aiohttp #3535 issue with SSL data after close
 
@@ -84,7 +96,8 @@ def ignore_aiohttp_ssl_eror(loop, aiohttpversion="3.5.4"):
 
 class SearchSingleConsumer(AsyncHttpConsumer):
     @wrapper_500
-    async def handle(self, body):
+    @with_cache
+    async def handle(self, body, cache):
         lookup_type = self.scope["url_route"]["kwargs"]["lookup_type"]
         lookup = self.scope["url_route"]["kwargs"]["lookup"]
         ignore_aiohttp_ssl_eror(asyncio.get_running_loop())
@@ -94,7 +107,7 @@ class SearchSingleConsumer(AsyncHttpConsumer):
             await json_response(self, 400, {"detail": "Invalid lookup"})
         try:
             async with aiohttp.client.ClientSession(timeout=aiohttp_timeout) as session:
-                data = await source.get(lookup, session, cache=cache.get_default())
+                data = await source.get(lookup, session, cache=cache)
             profile = sources.result_to_retribute_profile(lookup_type, lookup, data)
         except (exceptions.SearchError, aiohttp.ClientError) as e:
             await json_response(self, 400, {"detail": e.message})
@@ -105,9 +118,9 @@ class SearchSingleConsumer(AsyncHttpConsumer):
         await json_response(self, 200, profile)
 
 
-async def do_lookup(lookup, lookup_type, session, source, results):
+async def do_lookup(lookup, lookup_type, session, source, results, cache):
     try:
-        data = await source.get(lookup, session, cache=cache.get_default())
+        data = await source.get(lookup, session, cache=cache)
         profile = sources.result_to_retribute_profile(lookup_type, lookup, data)
     except (
         exceptions.SearchError,
@@ -126,7 +139,8 @@ async def do_lookup(lookup, lookup_type, session, source, results):
 
 class SearchMultipleConsumer(AsyncHttpConsumer):
     @wrapper_500
-    async def handle(self, body):
+    @with_cache
+    async def handle(self, body, cache):
         if self.scope["method"] == "OPTIONS":
             return await self.send_response(
                 200,
@@ -167,6 +181,7 @@ class SearchMultipleConsumer(AsyncHttpConsumer):
                         session=session,
                         source=source,
                         results=results,
+                        cache=cache,
                     )
                 )
             try: