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

Fix problems

Browse source
This commit is contained in:
Cyber MacGeddon 2025-04-29 00:06:08 +01:00
parent 47bb63607e
commit 5dfa125737
7 changed files with 18 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

2
trustgraph-base/trustgraph/base/consumer_spec.py
View file

@ -20,7 +20,7 @@ class ConsumerSpec(Spec):
flow = flow, flow = flow,
client = processor.pulsar_client, client = processor.pulsar_client,
topic = definition[self.name], topic = definition[self.name],
subscriber = processor.id + "--" + self.name, subscriber = processor.id + "--" + flow.name + "--" + self.name,
schema = self.schema, schema = self.schema,
handler = self.handler, handler = self.handler,
metrics = consumer_metrics, metrics = consumer_metrics,

3
trustgraph-base/trustgraph/base/flow_processor.py
View file

@ -75,7 +75,8 @@ class FlowProcessor(AsyncProcessor):
# Get list of flows which should be running and are currently # Get list of flows which should be running and are currently
# running # running
wanted_flows = flow_config.keys() wanted_flows = flow_config.keys()
current_flows = self.flows.keys() # This takes a copy, needed because dict gets modified by stop_flow
current_flows = list(self.flows.keys())
# Start all the flows which arent currently running # Start all the flows which arent currently running
for flow in wanted_flows: for flow in wanted_flows:

4
trustgraph-base/trustgraph/base/request_response_spec.py
View file

@ -127,7 +127,9 @@ class RequestResponseSpec(Spec):
rr = self.impl( rr = self.impl(
client = processor.pulsar_client, client = processor.pulsar_client,
subscription = flow.id, subscription = (
processor.id + "--" + flow.name + "--" + self.request_name
),
consumer_name = flow.id, consumer_name = flow.id,
request_topic = definition[self.request_name], request_topic = definition[self.request_name],
request_schema = self.request_schema, request_schema = self.request_schema,

2
trustgraph-base/trustgraph/base/subscriber.py
View file

@ -20,7 +20,7 @@ class Subscriber:
self.running = True self.running = True
self.metrics = metrics self.metrics = metrics
async def __del__(self): def __del__(self):
self.running = False self.running = False
async def start(self): async def start(self):

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

@ -27,6 +27,8 @@ class FlowConfig:
self.config["flow-classes"][msg.class_name] = msg.class_definition self.config["flow-classes"][msg.class_name] = msg.class_definition
self.config.version += 1
await self.config.push() await self.config.push()
return FlowResponse( return FlowResponse(
@ -39,6 +41,8 @@ class FlowConfig:
del self.config["flow-classes"][msg.class_name] del self.config["flow-classes"][msg.class_name]
self.config.version += 1
await self.config.push() await self.config.push()
return FlowResponse( return FlowResponse(
@ -135,6 +139,8 @@ class FlowConfig:
"interfaces": interfaces, "interfaces": interfaces,
}) })
self.config.version += 1
await self.config.push() await self.config.push()
return FlowResponse( return FlowResponse(
@ -186,6 +192,8 @@ class FlowConfig:
if msg.flow_id in self.config["flows"]: if msg.flow_id in self.config["flows"]:
del self.config["flows"][msg.flow_id] del self.config["flows"][msg.flow_id]
self.config.version += 1
await self.config.push() await self.config.push()
return FlowResponse( return FlowResponse(

2
trustgraph-flow/trustgraph/extract/kg/relationships/extract.py
View file

@ -173,7 +173,7 @@ class Processor(FlowProcessor):
o=Value(value=v.metadata.id, is_uri=True) o=Value(value=v.metadata.id, is_uri=True)
)) ))
await self.emit_edges( await self.emit_triples(
flow("triples"), flow("triples"),
Metadata( Metadata(
id=v.metadata.id, id=v.metadata.id,

6
trustgraph-flow/trustgraph/metering/counter.py
View file

@ -72,7 +72,7 @@ class Processor(FlowProcessor):
for k, v in config.items() for k, v in config.items()
} }
def get_prices(self, prices, modelname): def get_prices(self, modelname):
if modelname in self.prices: if modelname in self.prices:
model = self.prices[modelname] model = self.prices[modelname]
@ -90,9 +90,7 @@ class Processor(FlowProcessor):
__class__.input_token_metric.inc(num_in) __class__.input_token_metric.inc(num_in)
__class__.output_token_metric.inc(num_out) __class__.output_token_metric.inc(num_out)
model_input_price, model_output_price = self.get_prices( model_input_price, model_output_price = self.get_prices(modelname)
price_list, modelname
)
if model_input_price == None: if model_input_price == None:
cost_per_call = f"Model Not Found in Price list" cost_per_call = f"Model Not Found in Price list"