diff --git a/tests/test-get-config b/tests/test-get-config new file mode 100755 index 00000000..ecabe032 --- /dev/null +++ b/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) + diff --git a/trustgraph-base/trustgraph/clients/base.py b/trustgraph-base/trustgraph/clients/base.py index ac809123..25eac3b7 100644 --- a/trustgraph-base/trustgraph/clients/base.py +++ b/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, - logger=pulsar.ConsoleLogger(log_level), - authentication=auth, + pulsar_host, + logger=pulsar.ConsoleLogger(log_level), + authentication=auth, + listener=listener, ) else: self.client = pulsar.Client( - pulsar_host, - logger=pulsar.ConsoleLogger(log_level) + pulsar_host, + logger=pulsar.ConsoleLogger(log_level), + listener_name=listener, ) self.producer = self.client.create_producer( diff --git a/trustgraph-base/trustgraph/clients/config_client.py b/trustgraph-base/trustgraph/clients/config_client.py new file mode 100644 index 00000000..eb797071 --- /dev/null +++ b/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 +