From 793f986b06e9888e43d57589d6e86dbc0006e536 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Sun, 20 Apr 2025 09:15:12 +0100 Subject: [PATCH] Fix some typos --- trustgraph-base/trustgraph/base/consumer.py | 3 ++- trustgraph-base/trustgraph/base/consumer_spec.py | 2 +- .../trustgraph/storage/doc_embeddings/qdrant/write.py | 3 ++- .../trustgraph/storage/graph_embeddings/qdrant/write.py | 2 +- .../trustgraph/storage/triples/cassandra/write.py | 8 +++++--- 5 files changed, 11 insertions(+), 7 deletions(-) diff --git a/trustgraph-base/trustgraph/base/consumer.py b/trustgraph-base/trustgraph/base/consumer.py index d24e8f6f..ddedd93a 100644 --- a/trustgraph-base/trustgraph/base/consumer.py +++ b/trustgraph-base/trustgraph/base/consumer.py @@ -83,7 +83,8 @@ class Consumer: topic = self.topic, subscription_name = self.subscriber, schema = JsonSchema(self.schema), - initial_position = pos + initial_position = pos, + consumer_type = pulsar.ConsumerType.Shared, ) except Exception as e: diff --git a/trustgraph-base/trustgraph/base/consumer_spec.py b/trustgraph-base/trustgraph/base/consumer_spec.py index 3d01be8f..1bde250a 100644 --- a/trustgraph-base/trustgraph/base/consumer_spec.py +++ b/trustgraph-base/trustgraph/base/consumer_spec.py @@ -20,7 +20,7 @@ class ConsumerSpec(Spec): flow = flow, client = processor.client, topic = definition[self.name], - subscriber = flow.id, + subscriber = flow.id + "---" + self.name, schema = self.schema, handler = self.handler, metrics = consumer_metrics, diff --git a/trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py b/trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py index f92ce7de..f997abb6 100644 --- a/trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py +++ b/trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py @@ -43,7 +43,8 @@ class Processor(DocumentEmbeddingsStoreService): dim = len(vec) collection = ( - "d_" + message.metadata.user + "_" + message.metadata.collection + "_" + + "d_" + message.metadata.user + "_" + + message.metadata.collection + "_" + str(dim) ) diff --git a/trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py b/trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py index d8645f0a..ecefee4f 100755 --- a/trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py +++ b/trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py @@ -68,7 +68,7 @@ class Processor(GraphEmbeddingsStoreService): dim = len(vec) collection = self.get_collection( - dim, v.metadata.user, v.metadata.collection + dim, message.metadata.user, message.metadata.collection ) self.qdrant.upsert( diff --git a/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py b/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py index 10a78409..f8396692 100755 --- a/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py +++ b/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py @@ -50,13 +50,15 @@ class Processor(TriplesStoreService): if self.username and self.password: self.tg = TrustGraph( hosts=self.graph_host, - keyspace=v.metadata.user, table=v.metadata.collection, + keyspace=message.metadata.user, + table=message.metadata.collection, username=self.username, password=self.password ) else: self.tg = TrustGraph( hosts=self.graph_host, - keyspace=v.metadata.user, table=v.metadata.collection, + keyspace=message.metadata.user, + table=message.metadata.collection, ) except Exception as e: print("Exception", e, flush=True) @@ -65,7 +67,7 @@ class Processor(TriplesStoreService): self.table = table - for t in v.triples: + for t in message.triples: self.tg.insert( t.s.value, t.p.value,