Config service working through gateway

test-api/test-config-api

@@ -0,0 +1,174 @@
+#!/usr/bin/env python3
+
+import requests
+import json
+import sys
+
+url = "http://localhost:8088/api/v1/"
+
+############################################################################
+
+input = {
+    "operation": "config"
+}
+
+resp = requests.post(
+    f"{url}config",
+    json=input,
+)
+
+if resp.status_code != 200:
+    raise RuntimeError(f"Status code: {resp.status_code}")
+
+resp = resp.json()
+
+if "error" in resp:
+    print(f"Error: {resp['error']}")
+    sys.exit(1)
+
+print(json.dumps(resp, indent=4))
+
+############################################################################
+
+input = {
+    "operation": "put",
+    "type": "test",
+    "key": "key1",
+    "value": "value1"
+}
+
+resp = requests.post(
+    f"{url}config",
+    json=input,
+)
+
+if resp.status_code != 200:
+    raise RuntimeError(f"Status code: {resp.status_code}")
+
+resp = resp.json()
+
+if "error" in resp:
+    print(f"Error: {resp['error']}")
+    sys.exit(1)
+
+print(json.dumps(resp, indent=4))
+
+############################################################################
+
+input = {
+    "operation": "put",
+    "type": "test",
+    "key": "key2",
+    "value": "testing 1 2 3"
+}
+
+resp = requests.post(
+    f"{url}config",
+    json=input,
+)
+
+if resp.status_code != 200:
+    raise RuntimeError(f"Status code: {resp.status_code}")
+
+resp = resp.json()
+
+if "error" in resp:
+    print(f"Error: {resp['error']}")
+    sys.exit(1)
+
+print(json.dumps(resp, indent=4))
+
+############################################################################
+
+input = {
+    "operation": "get",
+    "type": "test",
+    "key": "key2",
+}
+
+resp = requests.post(
+    f"{url}config",
+    json=input,
+)
+
+if resp.status_code != 200:
+    raise RuntimeError(f"Status code: {resp.status_code}")
+
+resp = resp.json()
+
+if "error" in resp:
+    print(f"Error: {resp['error']}")
+    sys.exit(1)
+
+print(json.dumps(resp, indent=4))
+
+############################################################################
+
+input = {
+    "operation": "config"
+}
+
+resp = requests.post(
+    f"{url}config",
+    json=input,
+)
+
+if resp.status_code != 200:
+    raise RuntimeError(f"Status code: {resp.status_code}")
+
+resp = resp.json()
+
+if "error" in resp:
+    print(f"Error: {resp['error']}")
+    sys.exit(1)
+
+print(json.dumps(resp, indent=4))
+
+############################################################################
+
+input = {
+    "operation": "list",
+    "type": "test"
+}
+
+resp = requests.post(
+    f"{url}config",
+    json=input,
+)
+
+if resp.status_code != 200:
+    raise RuntimeError(f"Status code: {resp.status_code}")
+
+resp = resp.json()
+
+if "error" in resp:
+    print(f"Error: {resp['error']}")
+    sys.exit(1)
+
+print(json.dumps(resp, indent=4))
+
+############################################################################
+
+input = {
+    "operation": "getall",
+    "type": "test"
+}
+
+resp = requests.post(
+    f"{url}config",
+    json=input,
+)
+
+if resp.status_code != 200:
+    raise RuntimeError(f"Status code: {resp.status_code}")
+
+resp = resp.json()
+
+if "error" in resp:
+    print(f"Error: {resp['error']}")
+    sys.exit(1)
+
+print(json.dumps(resp, indent=4))
+
+############################################################################
+

trustgraph-base/trustgraph/schema/__init__.py

@@ -11,5 +11,6 @@ from . metadata import *
 from . agent import *
 from . lookup import *
 from . library import *
+from . config import *
 
 

trustgraph-base/trustgraph/schema/config.py

@@ -7,20 +7,14 @@ from . types import Error, RowSchema
 ############################################################################
 
 # Prompt services, abstract the prompt generation
-
+class ConfigRequest(Record):
-class ConfigItem(Record):
+    operation = String() # get, list, getall, delete, put, dump
+    type = String()
     key = String()
     value = String()
 
-class ConfigItems(Record):
-    items = Map(String())
-
-class ConfigRequest(Record):
-    type = String()
-    key = String()
-    operation = String() # get, list, getall, delete, put, dump
-
 class ConfigResponse(Record):
+    version = Integer()
     value = String()
     directory = Array(String())
     values = Map(String())
@@ -28,16 +22,17 @@ class ConfigResponse(Record):
     error = Error()
 
 class ConfigPush(Record):
-    config = Map(ConfigItems())
+    version = Integer()
+    config = Map(Map(String()))
 
 config_request_queue = topic(
-    'prompt', kind='non-persistent', namespace='request'
+    'config', kind='non-persistent', namespace='request'
 )
 config_response_queue = topic(
-    'prompt', kind='non-persistent', namespace='response'
+    'config', kind='non-persistent', namespace='response'
 )
 config_push_queue = topic(
-    'prompt', kind='non-persistent', namespace='config'b
+    'config', kind='persistent', namespace='config'
 )
 
 ############################################################################

trustgraph-flow/trustgraph/config/service/service.py

@@ -4,13 +4,14 @@ Config service.  Fetchs an extract from the Wikipedia page
 using the API.
 """
 
-from trustgraph.schema import ConfigItem, ConfigItems, ConfigUpdate
+from pulsar.schema import JsonSchema
+
 from trustgraph.schema import ConfigRequest, ConfigResponse, ConfigPush
+from trustgraph.schema import Error
 from trustgraph.schema import config_request_queue, config_response_queue
 from trustgraph.schema import config_push_queue
 from trustgraph.log_level import LogLevel
 from trustgraph.base import ConsumerProducer
-import requests
 
 module = ".".join(__name__.split(".")[1:-1])
 
@@ -27,7 +28,6 @@ class Processor(ConsumerProducer):
         output_queue = params.get("output_queue", default_output_queue)
         push_queue = params.get("push_queue", default_push_queue)
         subscriber = params.get("subscriber", default_subscriber)
-        url = params.get("url", default_url)
 
         super(Processor, self).__init__(
             **params | {
@@ -51,24 +51,32 @@ class Processor(ConsumerProducer):
         # back-end state store.
         self.config = {}
 
-    async def handle_get(self, v):
+        # Version counter
+        self.version = 0
+
+    async def start(self):
+        await self.push()
+        
+    async def handle_get(self, v, id):
 
         if v.type in self.config:
 
             if v.key in self.config[v.type]:
 
                 resp = ConfigResponse(
+                    version = self.version,
                     value = self.config[v.type][v.key],
                     directory = None,
                     values = None,
                     config = None,
                     error = None,
                 )
-                await self.send(r, properties={"id": id})
+                await self.send(resp, properties={"id": id})
 
             else:
 
                 resp = ConfigResponse(
+                    version = None,
                     value=None,
                     directory=None,
                     values=None,
@@ -78,11 +86,12 @@ class Processor(ConsumerProducer):
                         message="No such key"
                     )
                 )
-                await self.send(r, properties={"id": id})
+                await self.send(resp, properties={"id": id})
 
         else:
 
             resp = ConfigResponse(
+                version = None,
                 value=None,
                 directory=None,
                 values=None,
@@ -92,24 +101,26 @@ class Processor(ConsumerProducer):
                     message="No such type"
                 )
             )
-            await self.send(r, properties={"id": id})
+            await self.send(resp, properties={"id": id})
 
-    async def handle_list(self, v):
+    async def handle_list(self, v, id):
 
         if v.type in self.config:
 
             resp = ConfigResponse(
+                version = self.version,
                 value = None,
-                directory = list(self.config[v.type].keys())
+                directory = list(self.config[v.type].keys()),
                 values = None,
                 config = None,
                 error = None,
             )
-            await self.send(r, properties={"id": id})
+            await self.send(resp, properties={"id": id})
 
         else:
 
             resp = ConfigResponse(
+                version = None,
                 value=None,
                 directory=None,
                 values=None,
@@ -119,24 +130,26 @@ class Processor(ConsumerProducer):
                     message="No such type"
                 )
             )
-            await self.send(r, properties={"id": id})
+            await self.send(resp, properties={"id": id})
 
-    async def handle_getall(self, v):
+    async def handle_getall(self, v, id):
 
         if v.type in self.config:
 
             resp = ConfigResponse(
+                version = self.version,
                 value = None,
                 directory = None,
-                values = self.config[v.type]
+                values = self.config[v.type],
                 config = None,
                 error = None,
             )
-            await self.send(r, properties={"id": id})
+            await self.send(resp, properties={"id": id})
 
         else:
 
             resp = ConfigResponse(
+                version = None,
                 value=None,
                 directory=None,
                 values=None,
@@ -146,24 +159,32 @@ class Processor(ConsumerProducer):
                     message="No such type"
                 )
             )
-            await self.send(r, properties={"id": id})
+            await self.send(resp, properties={"id": id})
 
-    async def handle_delete(self, v):
+    async def handle_delete(self, v, id):
 
         if v.type in self.config:
+
             if v.key in self.config[v.type]:
 
+                del self.config[v.type][v.key]
+                self.version += 1
+
                 resp = ConfigResponse(
+                    version = None,
                     value = None,
                     directory = None,
                     values = None,
                     config = None,
                     error = None,
                 )
-                await self.send(r, properties={"id": id})
+                await self.send(resp, properties={"id": id})
+
+                await self.push()
                 return
 
         resp = ConfigResponse(
+            version = None,
             value=None,
             directory=None,
             values=None,
@@ -173,48 +194,53 @@ class Processor(ConsumerProducer):
                 message="No such object"
             )
         )
-        await self.send(r, properties={"id": id})
+        await self.send(resp, properties={"id": id})
 
         await self.push()
 
-    async def handle_put(self, v):
+    async def handle_put(self, v, id):
 
         if v.type not in self.config:
             self.config[v.type] = {}
 
         self.config[v.type][v.key] = v.value
+        self.version += 1
 
         resp = ConfigResponse(
+            version = None,
             value = None,
             directory = None,
             values = None,
             error = None,
         )
-        await self.send(r, properties={"id": id})
+        await self.send(resp, properties={"id": id})
 
         await self.push()
 
-    async def handle_dump(self, v):
+    async def handle_dump(self, v, id):
 
         resp = ConfigResponse(
+            version = self.version,
             value = None,
             directory = None,
             values = None,
             config = self.config,
             error = None,
         )
-        await self.send(r, properties={"id": id})
+        await self.send(resp, properties={"id": id})
 
     async def push(self):
 
-        resp = ConfigResponse(
+        resp = ConfigPush(
+            version = self.version,
             value = None,
             directory = None,
             values = None,
             config = self.config,
             error = None,
         )
-        self.push_prod.send(r)
+        self.push_prod.send(resp)
+        print("Pushed.")
         
     async def handle(self, msg):
 
@@ -229,27 +255,27 @@ class Processor(ConsumerProducer):
 
             if v.operation == "get":
 
-                self.handle_get(v, id)
+                await self.handle_get(v, id)
 
             elif v.operation == "list":
 
-                self.handle_list(v, id)
+                await self.handle_list(v, id)
 
             elif v.operation == "getall":
 
-                self.handle_getall(v, id)
+                await self.handle_getall(v, id)
 
             elif v.operation == "delete":
 
-                self.handle_delete(v, id)
+                await self.handle_delete(v, id)
 
             elif v.operation == "put":
 
-                self.handle_put(v, id)
+                await self.handle_put(v, id)
 
-            elif v.operation == "dump":
+            elif v.operation == "config":
 
-                self.handle_dump(v, id)
+                await self.handle_dump(v, id)
 
             else:
 
@@ -269,7 +295,7 @@ class Processor(ConsumerProducer):
 
         except Exception as e:
                 
-            r = LookupResponse(
+            r = ConfigResponse(
                 error=Error(
                     type = "unexpected-error",
                     message = str(e),

trustgraph-flow/trustgraph/gateway/service.py

@@ -38,6 +38,7 @@ from . agent import AgentRequestor
 from . dbpedia import DbpediaRequestor
 from . internet_search import InternetSearchRequestor
 from . librarian import LibrarianRequestor
+from . config import ConfigRequestor
 from . triples_stream import TriplesStreamEndpoint
 from . graph_embeddings_stream import GraphEmbeddingsStreamEndpoint
 from . document_embeddings_stream import DocumentEmbeddingsStreamEndpoint
@@ -141,6 +142,10 @@ class Api:
                 pulsar_client=self.pulsar_client, timeout=self.timeout,
                 auth = self.auth,
             ),
+            "config": ConfigRequestor(
+                pulsar_client=self.pulsar_client, timeout=self.timeout,
+                auth = self.auth,
+            ),
             "encyclopedia": EncyclopediaRequestor(
                 pulsar_client=self.pulsar_client, timeout=self.timeout,
                 auth = self.auth,
@@ -199,6 +204,10 @@ class Api:
                 endpoint_path = "/api/v1/librarian", auth=self.auth,
                 requestor = self.services["librarian"],
             ),
+            ServiceEndpoint(
+                endpoint_path = "/api/v1/config", auth=self.auth,
+                requestor = self.services["config"],
+            ),
             ServiceEndpoint(
                 endpoint_path = "/api/v1/encyclopedia", auth=self.auth,
                 requestor = self.services["encyclopedia"],