From cc17998944361d866c48bc9b7a287e78541ff5cb Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Tue, 1 Apr 2025 16:33:22 +0100 Subject: [PATCH] Config service working through gateway --- test-api/test-config-api | 174 ++++++++++++++++++ trustgraph-base/trustgraph/schema/__init__.py | 1 + trustgraph-base/trustgraph/schema/config.py | 23 +-- .../trustgraph/config/service/service.py | 90 +++++---- trustgraph-flow/trustgraph/gateway/service.py | 9 + 5 files changed, 251 insertions(+), 46 deletions(-) create mode 100755 test-api/test-config-api diff --git a/test-api/test-config-api b/test-api/test-config-api new file mode 100755 index 00000000..4ec6892a --- /dev/null +++ b/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)) + +############################################################################ + diff --git a/trustgraph-base/trustgraph/schema/__init__.py b/trustgraph-base/trustgraph/schema/__init__.py index 9c44a743..28e1a879 100644 --- a/trustgraph-base/trustgraph/schema/__init__.py +++ b/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 * diff --git a/trustgraph-base/trustgraph/schema/config.py b/trustgraph-base/trustgraph/schema/config.py index a95359fb..d2895d31 100644 --- a/trustgraph-base/trustgraph/schema/config.py +++ b/trustgraph-base/trustgraph/schema/config.py @@ -7,20 +7,14 @@ from . types import Error, RowSchema ############################################################################ # Prompt services, abstract the prompt generation - -class ConfigItem(Record): +class ConfigRequest(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' ) ############################################################################ diff --git a/trustgraph-flow/trustgraph/config/service/service.py b/trustgraph-flow/trustgraph/config/service/service.py index 3656e6dd..a2d27c7f 100644 --- a/trustgraph-flow/trustgraph/config/service/service.py +++ b/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), diff --git a/trustgraph-flow/trustgraph/gateway/service.py b/trustgraph-flow/trustgraph/gateway/service.py index d3122e3b..e997f83e 100755 --- a/trustgraph-flow/trustgraph/gateway/service.py +++ b/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"],