Config client

tests/test-get-config

@@ -0,0 +1,11 @@
+#!/usr/bin/env python3
+
+import pulsar
+from trustgraph.clients.config_client import ConfigClient
+
+cli = ConfigClient(pulsar_host="pulsar://localhost:6650")
+
+resp = cli.request_config()
+
+print(resp)
+

trustgraph-base/trustgraph/clients/base.py

@@ -28,6 +28,7 @@ class BaseClient:
             output_schema=None,
             pulsar_host="pulsar://pulsar:6650",
             pulsar_api_key=None,
+            listener=None,
     ):
 
         if input_queue == None: raise RuntimeError("Need input_queue")
@@ -41,14 +42,16 @@ class BaseClient:
         if pulsar_api_key:
             auth = pulsar.AuthenticationToken(pulsar_api_key)
             self.client = pulsar.Client(
-            pulsar_host,
+                pulsar_host,
-            logger=pulsar.ConsoleLogger(log_level),
+                logger=pulsar.ConsoleLogger(log_level),
-            authentication=auth,
+                authentication=auth,
+                listener=listener,
             )
         else:
             self.client = pulsar.Client(
-            pulsar_host,
+                pulsar_host,
-            logger=pulsar.ConsoleLogger(log_level)
+                logger=pulsar.ConsoleLogger(log_level),
+                listener_name=listener,
             )
 
         self.producer = self.client.create_producer(

trustgraph-base/trustgraph/clients/config_client.py

@@ -0,0 +1,161 @@
+
+import _pulsar
+import json
+import dataclasses
+
+from .. schema import ConfigRequest, ConfigResponse, ConfigKey, ConfigValue
+from .. schema import config_request_queue
+from .. schema import config_response_queue
+from . base import BaseClient
+
+# Ugly
+ERROR=_pulsar.LoggerLevel.Error
+WARN=_pulsar.LoggerLevel.Warn
+INFO=_pulsar.LoggerLevel.Info
+DEBUG=_pulsar.LoggerLevel.Debug
+
+@dataclasses.dataclass
+class Definition:
+    name: str
+    definition: str
+
+@dataclasses.dataclass
+class Relationship:
+    s: str
+    p: str
+    o: str
+    o_entity: str
+
+@dataclasses.dataclass
+class Topic:
+    name: str
+    definition: str
+
+class ConfigClient(BaseClient):
+
+    def __init__(
+            self, log_level=ERROR,
+            subscriber=None,
+            input_queue=None,
+            output_queue=None,
+            pulsar_host="pulsar://pulsar:6650",
+            listener=None,
+            pulsar_api_key=None,
+    ):
+
+        if input_queue == None:
+            input_queue = config_request_queue
+
+        if output_queue == None:
+            output_queue = config_response_queue
+
+        super(ConfigClient, self).__init__(
+            log_level=log_level,
+            subscriber=subscriber,
+            input_queue=input_queue,
+            output_queue=output_queue,
+            pulsar_host=pulsar_host,
+            pulsar_api_key=pulsar_api_key,
+            input_schema=ConfigRequest,
+            output_schema=ConfigResponse,
+            listener=listener,
+        )
+
+    def request_get(self, keys, timeout=300):
+
+        resp = self.call(
+            id=id,
+            operation="get",
+            keys=[
+                ConfigKey(
+                    type = k["type"],
+                    key = k["key"]
+                )
+                for k in keys
+            ],
+            timeout=timeout
+        )
+
+        return [
+            {
+                "type": v.type,
+                "key": v.key,
+                "value": v.value
+            }
+            for v in resp.values
+        ]
+
+    def request_list(self, type, timeout=300):
+
+        resp = self.call(
+            id=id,
+            operation="list",
+            type=type,
+            timeout=timeout
+        )
+
+        return resp.directory
+
+    def request_getvalues(self, type, timeout=300):
+
+        resp = self.call(
+            id=id,
+            operation="getvalues",
+            type=type,
+            timeout=timeout
+        )
+
+        return [
+            {
+                "type": v.type,
+                "key": v.key,
+                "value": v.value
+            }
+            for v in resp.values
+        ]
+
+    def request_delete(self, keys, timeout=300):
+
+        resp = self.call(
+            id=id,
+            operation="delete",
+            keys=[
+                ConfigKey(
+                    type = k["type"],
+                    key = k["key"]
+                )
+                for k in keys
+            ],
+            timeout=timeout
+        )
+
+        return None
+
+    def request_put(self, value, timeout=300):
+
+        resp = self.call(
+            id=id,
+            operation="put",
+            values=[
+                ConfigValue(
+                    type = k["type"],
+                    key = k["key"],
+                    value = k["value"]
+                )
+                for k in keys
+            ],
+            timeout=timeout
+        )
+
+        return None
+
+    def request_config(self, timeout=300):
+
+        resp = self.call(
+            id=id,
+            operation="config",
+            timeout=timeout
+        )
+
+        return resp.config
+