apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-05-19 20:35:13 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Feature/config service (#332)

Browse source 
Configuration service provides an API to change configuration. Complete configuration is pushed down a config queue so that users have a complete copy of config object.
This commit is contained in:
cybermaggedon 2025-04-01 19:47:05 +01:00 • committed by GitHub
parent 21bda863a7
commit fa09dc319e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 1002 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

13
trustgraph-base/trustgraph/clients/base.py
View file

@ -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(

161
trustgraph-base/trustgraph/clients/config_client.py Normal file
View file

@ -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