added redis adapter

python_repositories/adapters/__init__.py

@@ -0,0 +1,5 @@
+from .redis_adapter import RedisAdapter as RedisAdapter
+
+__all__ = [
+    "RedisAdapter",
+]

python_repositories/adapters/redis_adapter.py

@@ -0,0 +1,149 @@
+"""Definition of RedisAdapter class."""
+
+from __future__ import annotations
+
+import os
+import structlog
+
+import redis
+from redis.commands.json.path import Path as RedisPath
+
+from python_utils import check_env
+
+from python_repositories.interfaces import (
+    ContextAwareInterface,
+    ConnectionAwareInterface,
+)
+
+
+class RedisAdapter(
+    ContextAwareInterface,
+    ConnectionAwareInterface,
+):
+    """Redis adapter exposing basic CRUD functionality."""
+
+    uri_env_var_name: str = "REDIS_URI"
+    path: str = RedisPath.root_path()
+    encoding: str = "UTF-8"
+
+    def __init__(self) -> None:
+        # Setup logger
+        self.logger = structlog.get_logger(
+            self.__class__.__name__,
+        )
+        # Check environment variables
+        check_env(self.uri_env_var_name)
+        # Prepare internal variables
+        self._client: redis.Redis | None = None
+
+    def __enter__(self) -> RedisAdapter:
+        """Enter the context."""
+        self.connect()
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type | None,
+        exc_val: object | None,
+        exc_tb: object | None
+    ) -> None:
+        """Exit the context."""
+        ctx_info = {
+            'exc_type': exc_type,
+            'exc_val': exc_val,
+            'exc_tb': exc_tb
+        }
+        if any(
+            (
+                exc_type is not None,
+                exc_val is not None,
+                exc_tb is not None,
+            ),
+        ):
+            self.logger.error("Error while exiting context", **ctx_info)
+        self.disconnect()
+
+    def connect(self) -> None:
+        """Connect to the Redis server."""
+        # Prepare arguments
+        uri = str(os.getenv(self.uri_env_var_name))
+        # Connect client
+        client = redis.Redis.from_url(
+            url=uri,
+            socket_connect_timeout=10,
+        )
+        if not client.ping():
+            raise ConnectionError(f"Could not connect to Redis at {uri}")
+        # Persist client
+        self._client = client
+
+    def disconnect(self) -> None:
+        """Disconnect from the Redis server."""
+        # Close connection
+        if self._client is not None:
+            self._client.close()
+        # Reset client
+        self._client = None
+
+    @property
+    def is_connected(self) -> bool:
+        """Check if connected to Redis server."""
+        res = bool(isinstance(self._client, redis.Redis))
+        self.logger.debug(res)
+        return res
+
+    def _set(self, key: str, data: dict) -> None:
+        """Set a JSON object in Redis."""
+        # Check input
+        if not isinstance(key, str) or len(key) == 0:
+            raise ValueError("Key must be a non-empty string")
+        if not isinstance(data, dict) or len(data) == 0:
+            raise ValueError("Data must be a non-empty dictionary")
+        # Check connection
+        if not self.is_connected:
+            raise ConnectionError("Not connected to Redis")
+        # Set data
+        self._client.json().set(key, self.path, data)
+        self.logger.debug(f"Set {key} to {data}")
+
+    def _get(self, key: str) -> dict | None:
+        """Get a JSON object from Redis."""
+        # Check input
+        if not isinstance(key, str) or len(key) == 0:
+            raise ValueError("Key must be a non-empty string")
+        # Check connection
+        if not self.is_connected:
+            raise ConnectionError("Not connected to Redis")
+        # Get data
+        data = self._client.json().get(key)
+        self.logger.debug(f"Got {data} from {key}")
+        return data
+
+    def _delete(self, key: str) -> None:
+        """Delete data from Redis."""
+        # Check input
+        if not isinstance(key, str) or len(key) == 0:
+            raise ValueError("Key must be a non-empty string")
+        # Check connection
+        if not self.is_connected:
+            raise ConnectionError("Not connected to Redis")
+        # Delete data
+        self._client.json().delete(key)
+        self.logger.debug(f"Deleted {key}")
+
+    def _list_keys(self, pattern: str) -> list[str]:
+        """List keys in Redis matching a pattern."""
+        # Check input
+        if not isinstance(pattern, str) or len(pattern) == 0:
+            raise ValueError("Pattern must be a non-empty string")
+        # Check connection
+        if not self.is_connected:
+            raise ConnectionError("Not connected to Redis")
+        # List keys
+        keys: list[str] = [
+            key.decode(self.encoding)
+            for key
+            in self._client.keys(pattern)
+        ]
+        self.logger.debug(f"Got {keys} matching {pattern}")
+        return keys