From 3b74a37a2b1f367ab6d66f921fc3897cf06324e5 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Mon, 31 Mar 2025 23:32:15 +0100 Subject: [PATCH] Bare bones of a config service --- trustgraph-cli/scripts/tg-init-pulsar | 7 + trustgraph-flow/scripts/config-svc | 6 + trustgraph-flow/setup.py | 1 + .../trustgraph/config/service/__init__.py | 3 + .../trustgraph/config/service/__main__.py | 7 + .../trustgraph/config/service/service.py | 203 ++++++++++++++++++ 6 files changed, 227 insertions(+) create mode 100755 trustgraph-flow/scripts/config-svc create mode 100644 trustgraph-flow/trustgraph/config/service/__init__.py create mode 100644 trustgraph-flow/trustgraph/config/service/__main__.py create mode 100644 trustgraph-flow/trustgraph/config/service/service.py diff --git a/trustgraph-cli/scripts/tg-init-pulsar b/trustgraph-cli/scripts/tg-init-pulsar index 07fd31eb..98b5072f 100755 --- a/trustgraph-cli/scripts/tg-init-pulsar +++ b/trustgraph-cli/scripts/tg-init-pulsar @@ -79,6 +79,13 @@ def init(url, tenant="tg"): } }) + ensure_namespace(url, tenant, "config", { + "retention_policies": { + "retentionSizeInMB": 50, + "retentionTimeInMinutes": -1, + } + }) + def main(): parser = argparse.ArgumentParser( diff --git a/trustgraph-flow/scripts/config-svc b/trustgraph-flow/scripts/config-svc new file mode 100755 index 00000000..9debd391 --- /dev/null +++ b/trustgraph-flow/scripts/config-svc @@ -0,0 +1,6 @@ +#!/usr/bin/env python3 + +from trustgraph.config.service import run + +run() + diff --git a/trustgraph-flow/setup.py b/trustgraph-flow/setup.py index 4b39a327..a4a55575 100644 --- a/trustgraph-flow/setup.py +++ b/trustgraph-flow/setup.py @@ -73,6 +73,7 @@ setuptools.setup( "scripts/api-gateway", "scripts/chunker-recursive", "scripts/chunker-token", + "scripts/config-svc", "scripts/de-query-milvus", "scripts/de-query-pinecone", "scripts/de-query-qdrant", diff --git a/trustgraph-flow/trustgraph/config/service/__init__.py b/trustgraph-flow/trustgraph/config/service/__init__.py new file mode 100644 index 00000000..ba844705 --- /dev/null +++ b/trustgraph-flow/trustgraph/config/service/__init__.py @@ -0,0 +1,3 @@ + +from . service import * + diff --git a/trustgraph-flow/trustgraph/config/service/__main__.py b/trustgraph-flow/trustgraph/config/service/__main__.py new file mode 100644 index 00000000..e9136855 --- /dev/null +++ b/trustgraph-flow/trustgraph/config/service/__main__.py @@ -0,0 +1,7 @@ +#!/usr/bin/env python3 + +from . service import run + +if __name__ == '__main__': + run() + diff --git a/trustgraph-flow/trustgraph/config/service/service.py b/trustgraph-flow/trustgraph/config/service/service.py new file mode 100644 index 00000000..6812db51 --- /dev/null +++ b/trustgraph-flow/trustgraph/config/service/service.py @@ -0,0 +1,203 @@ + +""" +Config service. Fetchs an extract from the Wikipedia page +using the API. +""" + +from trustgraph.schema import ConfigItem, ConfigItems, ConfigUpdate +from trustgraph.schema import ConfigRequest, ConfigResponse, ConfigPush +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]) + +default_input_queue = config_request_queue +default_output_queue = config_response_queue +default_push_queue = config_push_queue +default_subscriber = module + +class Processor(ConsumerProducer): + + def __init__(self, **params): + + input_queue = params.get("input_queue", default_input_queue) + 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 | { + "input_queue": input_queue, + "output_queue": output_queue, + "push_queue": output_queue, + "subscriber": subscriber, + "input_schema": ConfigRequest, + "output_schema": ConfigResponse, + "push_schema": ConfigPush, + } + ) + + self.push_prod = self.client.create_producer( + topic=push_queue, + schema=JsonSchema(ConfigPush), + ) + + # FIXME: The state is held internally. This only works if there's + # one config service. Should be more than one, and use a + # back-end state store. + self.config = {} + + async def handle(self, msg): + + v = msg.value() + + # Sender-produced ID + id = msg.properties()["id"] + + print(f"Handling {id}...", flush=True) + + try: + + if v.operation == "get": + + if v.type in self.config: + if v.key in self.config[v.type]: + + resp = ConfigResponse( + value = self.config[v.type][v.key], + directory = None, + values = None, + error = None, + ) + await self.send(r, properties={"id": id}) + self.consumer.acknowledge(msg) + + else: + + resp = ConfigResponse( + value=None, + directory=None, + values=None, + error=Error( + code="no-such-key", + message="No such key" + ) + ) + await self.send(r, properties={"id": id}) + self.consumer.acknowledge(msg) + + else: + + resp = ConfigResponse( + value=None, + directory=None, + values=None, + error=Error( + code="no-such-type", + message="No such type" + ) + ) + await self.send(r, properties={"id": id}) + self.consumer.acknowledge(msg) + + elif v.operation == "list": + + if v.type in self.config: + + resp = ConfigResponse( + value = None, + directory = list(self.config[v.type].keys()) + values = None, + error = None, + ) + await self.send(r, properties={"id": id}) + self.consumer.acknowledge(msg) + + else: + + resp = ConfigResponse( + value=None, + directory=None, + values=None, + error=Error( + code="no-such-type", + message="No such type" + ) + ) + await self.send(r, properties={"id": id}) + self.consumer.acknowledge(msg) + + elif v.operation == "getall": + + if v.type in self.config: + + resp = ConfigResponse( + value = None, + directory = None, + values = self.config[v.type] + error = None, + ) + await self.send(r, properties={"id": id}) + self.consumer.acknowledge(msg) + + else: + + resp = ConfigResponse( + value=None, + directory=None, + values=None, + error=Error( + code="no-such-type", + message="No such type" + ) + ) + await self.send(r, properties={"id": id}) + self.consumer.acknowledge(msg) + + else: + r = ConfigResponse( + value=None, + directory=None, + values=None, + error=Error( + code="bad-operation", + message="Bad operation" + ) + ) + await self.send(r, properties={"id": id}) + self.consumer.acknowledge(msg) + + except Exception as e: + + r = LookupResponse( + error=Error( + type = "unexpected-error", + message = str(e), + ), + text=None, + ) + await self.send(r, properties={"id": id}) + self.consumer.acknowledge(msg) + + @staticmethod + def add_args(parser): + + ConsumerProducer.add_args( + parser, default_input_queue, default_subscriber, + default_output_queue, + ) + + parser.add_argument( + '-q', '--push-queue', + default=default_push_queue, + help=f'Config push queue (default: {default_push_queue})' + ) + +def run(): + + Processor.launch(module, __doc__) +