Best project in Budapest KIWI event

regiojet_search/cache.py

@@ -0,0 +1,61 @@
+import importlib
+import json
+from contextlib import contextmanager
+from typing import Optional
+
+import redis
+
+from regiojet_search import settings
+from regiojet_search.encoders import SearchResultJSONEncoder
+from regiojet_search.models import City, Result
+from regiojet_search.settings import CACHE_BACKEND, CACHE_SURNAME
+
+
+@contextmanager
+def get_redis_connection(host: str, port: int, db: int):
+    connection = redis.Redis(host=host, port=port, db=db)
+    yield connection
+    connection.close()
+
+
+class RedisCache:
+    def __init__(self, host: str = settings.CACHE_HOST, port: int = settings.CACHE_PORT, db: int = settings.CACHE_DB):
+        self.host = host
+        self.port = port
+        self.db = db
+
+    def get(self, key: str) -> Optional[list[dict]]:
+        with get_redis_connection(self.host, self.port, self.db) as redis_connection:
+            cached_value = redis_connection.get(f"{CACHE_SURNAME}:{key}")
+
+        try:
+            return json.loads(cached_value)
+        except Exception:  # noqa cache get is not that important, so make it fail-operational
+            return None
+
+    def set(self, key: str, value: list[City | Result]) -> None:
+        with get_redis_connection(self.host, self.port, self.db) as redis_connection:
+            redis_connection.set(f"{CACHE_SURNAME}:{key}", json.dumps(value, cls=SearchResultJSONEncoder))
+
+
+class DummyCache:
+    def __init__(self, host: str = settings.CACHE_HOST, port: int = settings.CACHE_PORT, db: int = settings.CACHE_DB):
+        self.host = host
+        self.port = port
+        self.db = db
+
+    def get(self, key: str) -> Optional[list[dict]]:
+        return None
+
+    def set(self, key: str, value: list[City | Result]) -> None:
+        return None
+
+
+def load_cache():
+    cache_backend_module = ".".join(CACHE_BACKEND.split(".")[:-1])
+    cache_backend_name = CACHE_BACKEND.split(".")[-1]
+    cache_class = getattr(importlib.import_module(cache_backend_module), cache_backend_name)
+    return cache_class()
+
+
+cache = load_cache()