apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-17 17:21:02 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Changed schema to add batching

Browse source
This commit is contained in:
Cyber MacGeddon 2025-04-01 19:06:29 +01:00
parent 1762ecaf90
commit 1b2de0c76f
4 changed files with 274 additions and 155 deletions
Download patch file Download diff file Expand all files Collapse all files

92
test-api/test-config-api
View file

@ -32,9 +32,18 @@ print(json.dumps(resp, indent=4))
input = {
"operation": "put",
"type": "test",
"key": "key1",
"value": "value1"
"values": [
{
"type": "test",
"key": "key1",
"value": "value1"
},
{
"type": "test",
"key": "key2",
"value": "value2"
}
]
}
resp = requests.post(
@ -57,9 +66,13 @@ print(json.dumps(resp, indent=4))
input = {
"operation": "put",
"type": "test",
"key": "key2",
"value": "testing 1 2 3"
"values": [
{
"type": "test",
"key": "key3",
"value": "testing 1 2 3"
}
]
}
resp = requests.post(
@ -82,8 +95,16 @@ print(json.dumps(resp, indent=4))
input = {
"operation": "get",
"type": "test",
"key": "key2",
"keys": [
{
"type": "test",
"key": "key2"
},
{
"type": "test",
"key": "key3"
}
]
}
resp = requests.post(
@ -150,7 +171,7 @@ print(json.dumps(resp, indent=4))
############################################################################
input = {
"operation": "getall",
"operation": "getvalues",
"type": "test"
}
@ -172,3 +193,56 @@ print(json.dumps(resp, indent=4))
############################################################################
input = {
"operation": "delete",
"keys": [
{
"type": "test",
"key": "key1"
},
{
"type": "test",
"key": "key3"
}
]
}
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))
############################################################################

42
trustgraph-base/trustgraph/schema/config.py
View file

@ -6,19 +6,51 @@ from . types import Error, RowSchema
############################################################################
# Prompt services, abstract the prompt generation
class ConfigRequest(Record):
operation = String() # get, list, getall, delete, put, dump
# Config service:
# get(keys) -> (version, values)
# list(type) -> (version, values)
# getvalues(type) -> (version, values)
# put(values) -> ()
# delete(keys) -> ()
# config() -> (version, config)
class ConfigKey(Record):
type = String()
key = String()
class ConfigValue(Record):
type = String()
key = String()
value = String()
# Prompt services, abstract the prompt generation
class ConfigRequest(Record):
operation = String() # get, list, getvalues, delete, put, config
# get, delete
keys = Array(ConfigKey())
# list, getvalues
type = String()
# put
values = Array(ConfigValue())
class ConfigResponse(Record):
# get, list, getvalues, config
version = Integer()
value = String()
# get, getvalues
values = Array(ConfigValue())
# list
directory = Array(String())
values = Map(String())
# config
config = Map(Map(String()))
# Everything
error = Error()
class ConfigPush(Record):

249
trustgraph-flow/trustgraph/config/service/service.py
View file

@ -7,7 +7,7 @@ using the API.
from pulsar.schema import JsonSchema
from trustgraph.schema import ConfigRequest, ConfigResponse, ConfigPush
from trustgraph.schema import Error
from trustgraph.schema import ConfigValue, Error
from trustgraph.schema import config_request_queue, config_response_queue
from trustgraph.schema import config_push_queue
from trustgraph.log_level import LogLevel
@ -20,6 +20,19 @@ default_output_queue = config_response_queue
default_push_queue = config_push_queue
default_subscriber = module
# This behaves just like a dict, should be easier to add persistent storage
# later
class ConfigurationItems(dict):
pass
class Configuration(dict):
def __getitem__(self, key):
if key not in self:
self[key] = ConfigurationItems()
return dict.__getitem__(self, key)
class Processor(ConsumerProducer):
def __init__(self, **params):
@ -49,7 +62,7 @@ class Processor(ConsumerProducer):
# 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 = {}
self.config = Configuration()
# Version counter
self.version = 0
@ -59,167 +72,142 @@ class Processor(ConsumerProducer):
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(resp, properties={"id": id})
else:
resp = ConfigResponse(
for k in v.keys:
if k.type not in self.config or k.key not in self.config[k.type]:
return ConfigResponse(
version = None,
value=None,
directory=None,
values=None,
values = None,
directory = None,
config = None,
error=Error(
code="no-such-key",
message="No such key"
error = Error(
code = "key-error",
message = f"Key error"
)
)
await self.send(resp, properties={"id": id})
else:
resp = ConfigResponse(
version = None,
value=None,
directory=None,
values=None,
config = None,
error=Error(
code="no-such-type",
message="No such type"
)
values = [
ConfigValue(
type = k.type,
key = k.key,
value = self.config[k.type][k.key]
)
await self.send(resp, properties={"id": id})
for k in v.keys
]
return ConfigResponse(
version = self.version,
values = values,
directory = None,
config = None,
error = None,
)
async def handle_list(self, v, id):
if v.type in self.config:
if v.type not in self.config:
resp = ConfigResponse(
version = self.version,
value = None,
directory = list(self.config[v.type].keys()),
return ConfigResponse(
version = None,
values = None,
config = None,
error = None,
)
await self.send(resp, properties={"id": id})
else:
resp = ConfigResponse(
version = None,
value=None,
directory=None,
values=None,
config = None,
error=Error(
code="no-such-type",
message="No such type"
)
)
await self.send(resp, properties={"id": id})
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],
config = None,
error = None,
error = Error(
code="key-error",
message="No such type",
),
)
await self.send(resp, properties={"id": id})
else:
return ConfigResponse(
version = self.version,
values = None,
directory = list(self.config[v.type].keys()),
config = None,
error = None,
)
resp = ConfigResponse(
async def handle_getvalues(self, v, id):
if v.type not in self.config:
return ConfigResponse(
version = None,
value=None,
directory=None,
values=None,
values = None,
directory = None,
config = None,
error=Error(
code="no-such-type",
message="No such type"
error = Error(
code = "key-error",
message = f"Key error"
)
)
await self.send(resp, properties={"id": id})
values = [
ConfigValue(
type = v.type,
key = k,
value = self.config[v.type][k],
)
for k in self.config[v.type]
]
return ConfigResponse(
version = self.version,
values = values,
directory = None,
config = None,
error = None,
)
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(
for k in v.keys:
if k.type not in self.config or k.key not in self.config[k.type]:
return ConfigResponse(
version = None,
value = None,
directory = None,
values = None,
directory = None,
config = None,
error = None,
error = Error(
code = "key-error",
message = f"Key error"
)
)
await self.send(resp, properties={"id": id})
await self.push()
return
for k in v.keys:
del self.config[k.type][k.key]
resp = ConfigResponse(
version = None,
value=None,
directory=None,
values=None,
config = None,
error=Error(
code="no-such-object",
message="No such object"
)
)
await self.send(resp, properties={"id": id})
self.version += 1
await self.push()
return ConfigResponse(
version = None,
value = None,
directory = None,
values = None,
config = None,
error = None,
)
async def handle_put(self, v, id):
if v.type not in self.config:
self.config[v.type] = {}
for k in v.values:
self.config[k.type][k.key] = k.value
self.config[v.type][v.key] = v.value
self.version += 1
resp = ConfigResponse(
await self.push()
return ConfigResponse(
version = None,
value = None,
directory = None,
values = None,
error = None,
)
await self.send(resp, properties={"id": id})
await self.push()
async def handle_config(self, v, id):
async def handle_dump(self, v, id):
resp = ConfigResponse(
return ConfigResponse(
version = self.version,
value = None,
directory = None,
@ -227,7 +215,6 @@ class Processor(ConsumerProducer):
config = self.config,
error = None,
)
await self.send(resp, properties={"id": id})
async def push(self):
@ -255,31 +242,31 @@ class Processor(ConsumerProducer):
if v.operation == "get":
await self.handle_get(v, id)
resp = await self.handle_get(v, id)
elif v.operation == "list":
await self.handle_list(v, id)
resp = await self.handle_list(v, id)
elif v.operation == "getall":
elif v.operation == "getvalues":
await self.handle_getall(v, id)
resp = await self.handle_getvalues(v, id)
elif v.operation == "delete":
await self.handle_delete(v, id)
resp = await self.handle_delete(v, id)
elif v.operation == "put":
await self.handle_put(v, id)
resp = await self.handle_put(v, id)
elif v.operation == "config":
await self.handle_dump(v, id)
resp = await self.handle_config(v, id)
else:
r = ConfigResponse(
resp = ConfigResponse(
value=None,
directory=None,
values=None,
@ -288,21 +275,21 @@ class Processor(ConsumerProducer):
message="Bad operation"
)
)
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
await self.send(resp, properties={"id": id})
self.consumer.acknowledge(msg)
except Exception as e:
r = ConfigResponse(
resp = ConfigResponse(
error=Error(
type = "unexpected-error",
message = str(e),
),
text=None,
)
await self.send(r, properties={"id": id})
await self.send(resp, properties={"id": id})
self.consumer.acknowledge(msg)
@staticmethod

46
trustgraph-flow/trustgraph/gateway/config.py
View file

@ -1,5 +1,5 @@
from .. schema import ConfigRequest, ConfigResponse
from .. schema import ConfigRequest, ConfigResponse, ConfigKey, ConfigValue
from .. schema import config_request_queue
from .. schema import config_response_queue
@ -20,30 +20,56 @@ class ConfigRequestor(ServiceRequestor):
def to_request(self, body):
if "keys" in body:
keys = [
ConfigKey(
type = k["type"],
key = k["key"],
)
for k in body["keys"]
]
else:
keys = None
if "values" in body:
values = [
ConfigValue(
type = v["type"],
key = v["key"],
value = v["value"],
)
for v in body["values"]
]
else:
values = None
return ConfigRequest(
operation = body.get("operation", None),
keys = keys,
type = body.get("type", None),
key = body.get("key", None),
value = body.get("value", None),
values = values
)
def from_response(self, message):
response = {
}
response = { }
if message.version:
response["version"] = message.version
if message.value:
response["value"] = message.value
if message.values:
response["values"] = [
{
"type": v.type,
"key": v.key,
"value": v.value,
}
for v in message.values
]
if message.directory:
response["directory"] = message.directory
if message.values:
response["values"] = message.values
if message.config:
response["config"] = message.config