Unification of pagination, admins control and more

misc/redis.py

@@ -0,0 +1,27 @@
+import logging
+from typing import Optional
+
+from redis.asyncio import Redis
+
+logger = logging.getLogger(__name__)
+
+
+class RedisClient:
+    def __init__(self, r: Redis):
+        self.r = r
+
+    async def set_search_query(self, query_hash: str, q: str, ttl: int = 3600) -> bool:
+        try:
+            await self.r.set(f"search:{query_hash}", q, ex=ttl)
+            return True
+        except Exception as e:
+            logger.exception(e)
+            return False
+
+    async def get_search_query(self, query_hash: str) -> Optional[str]:
+        try:
+            res = await self.r.get(f"search:{query_hash}")
+            return res.decode()
+        except Exception as e:
+            logger.exception(e)
+            return None