From 3277ddc63f30a6f46daae0a21adf90ecc6bddc95 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Mon, 29 Sep 2025 23:15:36 +0100 Subject: [PATCH] Fixing collection deletion --- .../trustgraph/storage/doc_embeddings/milvus/write.py | 6 ++++++ .../trustgraph/storage/doc_embeddings/pinecone/write.py | 6 ++++++ .../trustgraph/storage/doc_embeddings/qdrant/write.py | 8 ++++++++ .../trustgraph/storage/graph_embeddings/milvus/write.py | 6 ++++++ .../trustgraph/storage/graph_embeddings/pinecone/write.py | 6 ++++++ .../trustgraph/storage/graph_embeddings/qdrant/write.py | 8 ++++++++ .../trustgraph/storage/objects/cassandra/write.py | 8 +++++++- .../trustgraph/storage/triples/cassandra/write.py | 6 ++++++ .../trustgraph/storage/triples/falkordb/write.py | 6 ++++++ .../trustgraph/storage/triples/memgraph/write.py | 6 ++++++ trustgraph-flow/trustgraph/storage/triples/neo4j/write.py | 6 ++++++ 11 files changed, 71 insertions(+), 1 deletion(-) diff --git a/trustgraph-flow/trustgraph/storage/doc_embeddings/milvus/write.py b/trustgraph-flow/trustgraph/storage/doc_embeddings/milvus/write.py index 598183f2..2827b9b5 100755 --- a/trustgraph-flow/trustgraph/storage/doc_embeddings/milvus/write.py +++ b/trustgraph-flow/trustgraph/storage/doc_embeddings/milvus/write.py @@ -60,6 +60,12 @@ class Processor(DocumentEmbeddingsStoreService): metrics=storage_response_metrics, ) + async def start(self): + """Start the processor and its storage management consumer""" + await super().start() + await self.storage_request_consumer.start() + await self.storage_response_producer.start() + async def store_document_embeddings(self, message): for emb in message.chunks: diff --git a/trustgraph-flow/trustgraph/storage/doc_embeddings/pinecone/write.py b/trustgraph-flow/trustgraph/storage/doc_embeddings/pinecone/write.py index a613320a..e3b32ea3 100644 --- a/trustgraph-flow/trustgraph/storage/doc_embeddings/pinecone/write.py +++ b/trustgraph-flow/trustgraph/storage/doc_embeddings/pinecone/write.py @@ -115,6 +115,12 @@ class Processor(DocumentEmbeddingsStoreService): "Gave up waiting for index creation" ) + async def start(self): + """Start the processor and its storage management consumer""" + await super().start() + await self.storage_request_consumer.start() + await self.storage_response_producer.start() + async def store_document_embeddings(self, message): for emb in message.chunks: diff --git a/trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py b/trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py index 8f393b1a..2a13dc32 100644 --- a/trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py +++ b/trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py @@ -71,6 +71,14 @@ class Processor(DocumentEmbeddingsStoreService): metrics=storage_response_metrics, ) + async def start(self): + """Start the processor and its storage management consumer""" + await super().start() + if hasattr(self, 'storage_request_consumer'): + await self.storage_request_consumer.start() + if hasattr(self, 'storage_response_producer'): + await self.storage_response_producer.start() + async def store_document_embeddings(self, message): for emb in message.chunks: diff --git a/trustgraph-flow/trustgraph/storage/graph_embeddings/milvus/write.py b/trustgraph-flow/trustgraph/storage/graph_embeddings/milvus/write.py index f94f2752..ce7287fb 100755 --- a/trustgraph-flow/trustgraph/storage/graph_embeddings/milvus/write.py +++ b/trustgraph-flow/trustgraph/storage/graph_embeddings/milvus/write.py @@ -60,6 +60,12 @@ class Processor(GraphEmbeddingsStoreService): metrics=storage_response_metrics, ) + async def start(self): + """Start the processor and its storage management consumer""" + await super().start() + await self.storage_request_consumer.start() + await self.storage_response_producer.start() + async def store_graph_embeddings(self, message): for entity in message.entities: diff --git a/trustgraph-flow/trustgraph/storage/graph_embeddings/pinecone/write.py b/trustgraph-flow/trustgraph/storage/graph_embeddings/pinecone/write.py index b4d9ac5e..d2dd0223 100755 --- a/trustgraph-flow/trustgraph/storage/graph_embeddings/pinecone/write.py +++ b/trustgraph-flow/trustgraph/storage/graph_embeddings/pinecone/write.py @@ -115,6 +115,12 @@ class Processor(GraphEmbeddingsStoreService): "Gave up waiting for index creation" ) + async def start(self): + """Start the processor and its storage management consumer""" + await super().start() + await self.storage_request_consumer.start() + await self.storage_response_producer.start() + async def store_graph_embeddings(self, message): for entity in message.entities: diff --git a/trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py b/trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py index 2b67adf7..f013a839 100755 --- a/trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py +++ b/trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py @@ -96,6 +96,14 @@ class Processor(GraphEmbeddingsStoreService): return cname + async def start(self): + """Start the processor and its storage management consumer""" + await super().start() + if hasattr(self, 'storage_request_consumer'): + await self.storage_request_consumer.start() + if hasattr(self, 'storage_response_producer'): + await self.storage_response_producer.start() + async def store_graph_embeddings(self, message): for entity in message.entities: diff --git a/trustgraph-flow/trustgraph/storage/objects/cassandra/write.py b/trustgraph-flow/trustgraph/storage/objects/cassandra/write.py index 2ec98711..2350c161 100644 --- a/trustgraph-flow/trustgraph/storage/objects/cassandra/write.py +++ b/trustgraph-flow/trustgraph/storage/objects/cassandra/write.py @@ -340,6 +340,12 @@ class Processor(FlowProcessor): logger.warning(f"Failed to convert value {value} to type {field_type}: {e}") return str(value) + async def start(self): + """Start the processor and its storage management consumer""" + await super().start() + await self.storage_request_consumer.start() + await self.storage_response_producer.start() + async def on_object(self, msg, consumer, flow): """Process incoming ExtractedObject and store in Cassandra""" @@ -459,7 +465,7 @@ class Processor(FlowProcessor): message=str(e) ) ) - await self.send("storage-response", response) + await self.storage_response_producer.send(response) async def delete_collection(self, user: str, collection: str): """Delete all data for a specific collection""" diff --git a/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py b/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py index e925ece0..f057aed2 100755 --- a/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py +++ b/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py @@ -117,6 +117,12 @@ class Processor(TriplesStoreService): t.o.value ) + async def start(self): + """Start the processor and its storage management consumer""" + await super().start() + await self.storage_request_consumer.start() + await self.storage_response_producer.start() + async def on_storage_management(self, message): """Handle storage management requests""" logger.info(f"Storage management request: {message.operation} for {message.user}/{message.collection}") diff --git a/trustgraph-flow/trustgraph/storage/triples/falkordb/write.py b/trustgraph-flow/trustgraph/storage/triples/falkordb/write.py index 6591bafc..c9b371c2 100755 --- a/trustgraph-flow/trustgraph/storage/triples/falkordb/write.py +++ b/trustgraph-flow/trustgraph/storage/triples/falkordb/write.py @@ -185,6 +185,12 @@ class Processor(TriplesStoreService): help=f'FalkorDB database (default: {default_database})' ) + async def start(self): + """Start the processor and its storage management consumer""" + await super().start() + await self.storage_request_consumer.start() + await self.storage_response_producer.start() + async def on_storage_management(self, message): """Handle storage management requests""" logger.info(f"Storage management request: {message.operation} for {message.user}/{message.collection}") diff --git a/trustgraph-flow/trustgraph/storage/triples/memgraph/write.py b/trustgraph-flow/trustgraph/storage/triples/memgraph/write.py index 04f01f3d..61b2f5e5 100755 --- a/trustgraph-flow/trustgraph/storage/triples/memgraph/write.py +++ b/trustgraph-flow/trustgraph/storage/triples/memgraph/write.py @@ -317,6 +317,12 @@ class Processor(TriplesStoreService): help=f'Memgraph database (default: {default_database})' ) + async def start(self): + """Start the processor and its storage management consumer""" + await super().start() + await self.storage_request_consumer.start() + await self.storage_response_producer.start() + async def on_storage_management(self, message): """Handle storage management requests""" logger.info(f"Storage management request: {message.operation} for {message.user}/{message.collection}") diff --git a/trustgraph-flow/trustgraph/storage/triples/neo4j/write.py b/trustgraph-flow/trustgraph/storage/triples/neo4j/write.py index a59f9a7e..9f8ce7e2 100755 --- a/trustgraph-flow/trustgraph/storage/triples/neo4j/write.py +++ b/trustgraph-flow/trustgraph/storage/triples/neo4j/write.py @@ -268,6 +268,12 @@ class Processor(TriplesStoreService): help=f'Neo4j database (default: {default_database})' ) + async def start(self): + """Start the processor and its storage management consumer""" + await super().start() + await self.storage_request_consumer.start() + await self.storage_response_producer.start() + async def on_storage_management(self, message): """Handle storage management requests""" logger.info(f"Storage management request: {message.operation} for {message.user}/{message.collection}")