Repository: SINTEF/dlite (language: python)

Issue:
Add plugin for redis
To support fast in-memory exchange of instances.


Discussion / hints:

Fix patch:
diff --git a/storages/python/python-storage-plugins/redis.py b/storages/python/python-storage-plugins/redis.py
new file mode 100644
index 000000000..36a44b0d5
--- /dev/null
+++ b/storages/python/python-storage-plugins/redis.py
@@ -0,0 +1,123 @@
+"""A simple demonstrage of a DLite storage plugin written in Python."""
+from redis import Redis
+
+import dlite
+from dlite.options import Options
+
+
+class redis(dlite.DLiteStorageBase):
+    """DLite storage plugin for YAML."""
+
+    def open(self, uri: str, options=None):
+        """Opens `uri`.
+
+        Arguments:
+            uri: A fully resolved URI to the PostgreSQL database.
+            options: Supported options:
+            - `port`: Port to connect to (default: 6379).
+            - `username`: Redis user name.
+            - `password`: Redis password.
+            - `ssl`: Whether to ssl-encrypt the connection (default: false).
+            - `ssl_certfile`: Path to SSL certificate signing request (.crt).
+            - `ssl_keyfile`: Path to SSL private key file (.key).
+            - `ssl_ca_certs`: Path to SSL certificate container (.pem).
+            - `db`: Database number (default: 0).
+            - `expire`: Number of seconds before new keys expire.
+            - `only_new`: Wheter to only store instances that are not already
+              in the storage.
+            - `only_update`: Wheter to only update existing instances.
+            - `fernet_key`: A base64-encoded 32-byte Fernet key.  If given,
+              transparently encrypt all instances before sending them to Redis.
+              Generate the key with `crystallography.fernet.generate_key()`.
+        """
+        self.uri = uri
+        opts = Options(options, defaults="port=6379;ssl=false;db=0")
+
+        # Pop out options passed to redis.set()
+        self.setopts = {
+            "nx": dlite.asbool(opts.pop("only_new", "false")),
+            "xx": dlite.asbool(opts.pop("only_update", "false")),
+        }
+        if "expire" in opts:
+            self.setopts["ex"] = int(opts.pop("expire"))
+
+        self.fernet_key = opts.pop("fernet_key", None)
+
+        # Fix option types and open Redis connection
+        opts.port = int(opts.port)
+        opts.ssl = dlite.asbool(opts.ssl)
+        opts.db = int(opts.db)
+        self.redis = Redis(host=uri, **opts)
+        self.closed = False
+
+    def close(self):
+        """Close the connection to Redis."""
+        self.redis.close()
+        self.closed = True
+
+    def load(self, id: str):
+        """Loads `id` from current storage and return it as a new instance.
+
+        Arguments:
+            id: URI or UUID of DLite Instance to return.
+
+        Returns:
+            A DLite Instance corresponding to the given `id`.
+        """
+        uuid = dlite.get_uuid(id)
+        data = self.redis.get(f"dlite:{uuid}")
+        if data is None:
+            raise dlite.DLiteError(f"No such instance redis storage: {uuid}")
+        if self.fernet_key:
+            from cryptography.fernet import Fernet
+            f = Fernet(self.fernet_key.encode())
+            data = f.decrypt(data)
+        return dlite.Instance.from_bson(data)
+
+    def save(self, inst: dlite.Instance):
+        """Stores `inst` in current storage.
+
+        Arguments:
+            inst: A DLite Instance to store in the storage.
+        """
+        data = bytes(inst.asbson())
+        if self.fernet_key:
+            from cryptography.fernet import Fernet
+            key = self.fernet_key.encode()
+            f = Fernet(self.fernet_key)
+            data = f.encrypt(data)
+        self.redis.set(f"dlite:{inst.uuid}", data, **self.setopts)
+
+    def delete(self, id):
+        """Delete instance with given `id` from storage.
+
+        Arguments:
+            id: URI or UUID of instance to delete.
+        """
+        uuid = dlite.get_uuid(id)
+        self.redis.delete(uuid)
+
+    def queue(self, pattern=None):
+        """Generator method that iterates over all UUIDs in the storage
+        who"s metadata URI matches glob pattern `pattern`.
+
+        Arguments:
+            pattern: A glob pattern to filter the yielded UUIDs.
+
+        Yields:
+            DLite Instance UUIDs based on `pattern`.
+            If no `pattern` is given, all UUIDs are yielded from within the
+            storage.
+
+        Note:
+            This method fetch all instances in the Redis store and may
+            be slow on large databases.
+        """
+        for uuid in self.redis.keys("dlite:*"):
+            if pattern:
+                # Consider to make pattern more efficient by storing
+                # `uuid -> meta` mappings in redis as well...
+                inst = self.load(uuid)
+                if not dlite.globmatch(pattern, inst.meta.uri):
+                    continue
+            yield uuid


Test patch:
diff --git a/storages/python/tests-python/CMakeLists.txt b/storages/python/tests-python/CMakeLists.txt
index 47233ed06..375db9fe6 100644
--- a/storages/python/tests-python/CMakeLists.txt
+++ b/storages/python/tests-python/CMakeLists.txt
@@ -7,6 +7,7 @@ set(python-tests
   test_yaml_storage_python
   test_mongodb_python
   test_mongodb-atlas_python
+  test_redis
   )
 
 include(FindPythonModule)
diff --git a/storages/python/tests-python/test_redis.py b/storages/python/tests-python/test_redis.py
new file mode 100644
index 000000000..deb043523
--- /dev/null
+++ b/storages/python/tests-python/test_redis.py
@@ -0,0 +1,88 @@
+"""Script to test the 'redis' DLite plugin from Python."""
+import sys
+from time import sleep
+
+from pathlib import Path
+
+import dlite
+
+try:
+    import redis
+except ImportError:
+    print("redis-py not installed, skipping test")
+    sys.exit(44)  # skip test
+
+
+thisdir = Path(__file__).resolve().parent
+dlite.storage_path.append(thisdir / "input/*.json")
+
+
+inst1 = dlite.get_instance("2f8ba28c-add6-5718-a03c-ea46961d6ca7")
+inst2 = dlite.get_instance("52522ba5-6bfe-4a64-992d-e9ec4080fbac")
+meta = inst1.meta
+
+save0 = meta.asdict()
+save1 = inst1.asdict()
+save2 = inst2.asdict()
+
+
+# Test storing to Redis
+with dlite.Storage("redis", "localhost", "port=6379;expire=1;db=13") as s:
+    try:
+        s.save(inst1)
+    except dlite.DLiteError as exc:
+        if "ConnectionError:" in str(exc):
+            print(
+                f"cannot connect to Redis server: {exc}\n\n"
+                "Is the Redis server running?"
+            )
+            sys.exit(44)
+        raise
+    s.save(inst2)
+    s.save(inst1.meta)  # Save metadata as well
+
+
+# Test retrieving from Redis
+del meta
+del inst1
+del inst2
+with dlite.Storage("redis", "localhost", "port=6379;db=13") as s:
+    newmeta = s.load(save0["uuid"])
+    newinst1 = s.load(save1["uuid"])
+
+newinst2 = dlite.Instance.from_location(
+    "redis", "localhost", "port=6379;db=13", id=save2["uuid"]
+)
+
+
+# Test instance expiration
+print("Wait for instances in redis storage has expired...")
+sleep(1.1)
+
+del newinst1
+try:
+    with dlite.silent:
+        newinst1 = dlite.Instance.from_location(
+            "redis", "localhost", "port=6379;db=13", id=save1["uuid"]
+        )
+except dlite.DLiteError:
+    pass
+else:
+    raise RuntimeError("expected redis storage to have expired")
+
+
+# Test encryption
+try:
+    from cryptography.fernet import Fernet
+except ImportError:
+    print("Skipping testing encryption")
+else:
+    fernet_key = Fernet.generate_key().decode()
+    options = f"port=6379;expire=1;db=13;fernet_key={fernet_key}"
+
+    newinst2.save("redis", "localhost", options)
+
+    del newinst2
+    inst2 = dlite.Instance.from_location(
+        "redis", "localhost", options, id=save2["uuid"],
+    )