From 9f0de3794aa7909cffe016426e2bc7540066f17e Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Sun, 4 May 2025 21:44:03 +0100 Subject: [PATCH] Add/remove processing on table --- .../trustgraph/librarian/table_store.py | 52 +++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/trustgraph-flow/trustgraph/librarian/table_store.py b/trustgraph-flow/trustgraph/librarian/table_store.py index 73d39c79..47f3cac7 100644 --- a/trustgraph-flow/trustgraph/librarian/table_store.py +++ b/trustgraph-flow/trustgraph/librarian/table_store.py @@ -553,6 +553,58 @@ class TableStore: raise RuntimeError("No such document row?") + async def add_processing(self, processing): + + print("Adding processing", processing.id) + + while True: + + try: + + resp = self.cassandra.execute( + self.insert_processing_stmt, + ( + processing.id, processing.document_id, + processing.time, processing.flow, processing.user, + processing.collection, processing.tags + ) + ) + + break + + except Exception as e: + + print("Exception:", type(e)) + print(f"{e}, retry...", flush=True) + await asyncio.sleep(1) + + print("Add complete", flush=True) + + async def remove_processing(self, user, processing_id): + + print("Removing processing", processing_id) + + while True: + + try: + + resp = self.cassandra.execute( + self.delete_processing_stmt, + ( + user, processing_id + ) + ) + + break + + except Exception as e: + + print("Exception:", type(e)) + print(f"{e}, retry...", flush=True) + await asyncio.sleep(1) + + print("Delete complete", flush=True) + async def add_graph_embeddings(self, m): when = int(time.time() * 1000)