From 05a5127b64d04b81d36e90d30c9570047a3ae4c9 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Wed, 30 Jul 2025 23:12:52 +0100 Subject: [PATCH] Logging strategy updates --- .../decoding/mistral_ocr/processor.py | 24 +++-- .../direct/milvus_doc_embeddings.py | 11 ++- .../direct/milvus_graph_embeddings.py | 11 ++- .../direct/milvus_object_embeddings.py | 11 ++- .../document_embeddings/embeddings.py | 10 +- .../embeddings/graph_embeddings/embeddings.py | 10 +- .../trustgraph/external/wikipedia/service.py | 5 +- .../trustgraph/processing/processing.py | 23 +++-- trustgraph-flow/trustgraph/tables/config.py | 47 ++++------ .../trustgraph/tables/knowledge.py | 65 ++++++------- trustgraph-flow/trustgraph/tables/library.py | 91 ++++++++----------- 11 files changed, 148 insertions(+), 160 deletions(-) diff --git a/trustgraph-flow/trustgraph/decoding/mistral_ocr/processor.py b/trustgraph-flow/trustgraph/decoding/mistral_ocr/processor.py index e42d1601..4bacd278 100755 --- a/trustgraph-flow/trustgraph/decoding/mistral_ocr/processor.py +++ b/trustgraph-flow/trustgraph/decoding/mistral_ocr/processor.py @@ -19,6 +19,10 @@ from ... schema import document_ingest_queue, text_ingest_queue from ... log_level import LogLevel from ... base import InputOutputProcessor +import logging + +logger = logging.getLogger(__name__) + module = "ocr" default_subscriber = module @@ -94,18 +98,18 @@ class Processor(InputOutputProcessor): # Used with Mistral doc upload self.unique_id = str(uuid.uuid4()) - print("PDF inited") + logger.info("PDF inited") def ocr(self, blob): - print("Parse PDF...", flush=True) + logger.debug("Parse PDF...") pdfbuf = BytesIO(blob) pdf = PdfReader(pdfbuf) for chunk in chunks(pdf.pages, pages_per_chunk): - print("Get next pages...", flush=True) + logger.debug("Get next pages...") part = PdfWriter() for page in chunk: @@ -114,7 +118,7 @@ class Processor(InputOutputProcessor): buf = BytesIO() part.write_stream(buf) - print("Upload chunk...", flush=True) + logger.debug("Upload chunk...") uploaded_file = self.mistral.files.upload( file={ @@ -128,7 +132,7 @@ class Processor(InputOutputProcessor): file_id=uploaded_file.id, expiry=1 ) - print("OCR...", flush=True) + logger.debug("OCR...") processed = self.mistral.ocr.process( model="mistral-ocr-latest", @@ -139,21 +143,21 @@ class Processor(InputOutputProcessor): } ) - print("Extract markdown...", flush=True) + logger.debug("Extract markdown...") markdown = get_combined_markdown(processed) - print("OCR complete.", flush=True) + logger.info("OCR complete.") return markdown async def on_message(self, msg, consumer): - print("PDF message received") + logger.debug("PDF message received") v = msg.value() - print(f"Decoding {v.metadata.id}...", flush=True) + logger.info(f"Decoding {v.metadata.id}...") markdown = self.ocr(base64.b64decode(v.data)) @@ -164,7 +168,7 @@ class Processor(InputOutputProcessor): await consumer.q.output.send(r) - print("Done.", flush=True) + logger.info("Done.") @staticmethod def add_args(parser): diff --git a/trustgraph-flow/trustgraph/direct/milvus_doc_embeddings.py b/trustgraph-flow/trustgraph/direct/milvus_doc_embeddings.py index 9904f6ce..6d203858 100644 --- a/trustgraph-flow/trustgraph/direct/milvus_doc_embeddings.py +++ b/trustgraph-flow/trustgraph/direct/milvus_doc_embeddings.py @@ -1,6 +1,9 @@ from pymilvus import MilvusClient, CollectionSchema, FieldSchema, DataType import time +import logging + +logger = logging.getLogger(__name__) class DocVectors: @@ -21,7 +24,7 @@ class DocVectors: # Next time to reload - this forces a reload at next window self.next_reload = time.time() + self.reload_time - print("Reload at", self.next_reload) + logger.debug(f"Reload at {self.next_reload}") def init_collection(self, dimension): @@ -110,12 +113,12 @@ class DocVectors: } } - print("Loading...") + logger.debug("Loading...") self.client.load_collection( collection_name=coll, ) - print("Searching...") + logger.debug("Searching...") res = self.client.search( collection_name=coll, @@ -128,7 +131,7 @@ class DocVectors: # If reload time has passed, unload collection if time.time() > self.next_reload: - print("Unloading, reload at", self.next_reload) + logger.debug(f"Unloading, reload at {self.next_reload}") self.client.release_collection( collection_name=coll, ) diff --git a/trustgraph-flow/trustgraph/direct/milvus_graph_embeddings.py b/trustgraph-flow/trustgraph/direct/milvus_graph_embeddings.py index ce81a212..99cfb0b4 100644 --- a/trustgraph-flow/trustgraph/direct/milvus_graph_embeddings.py +++ b/trustgraph-flow/trustgraph/direct/milvus_graph_embeddings.py @@ -1,6 +1,9 @@ from pymilvus import MilvusClient, CollectionSchema, FieldSchema, DataType import time +import logging + +logger = logging.getLogger(__name__) class EntityVectors: @@ -21,7 +24,7 @@ class EntityVectors: # Next time to reload - this forces a reload at next window self.next_reload = time.time() + self.reload_time - print("Reload at", self.next_reload) + logger.debug(f"Reload at {self.next_reload}") def init_collection(self, dimension): @@ -110,12 +113,12 @@ class EntityVectors: } } - print("Loading...") + logger.debug("Loading...") self.client.load_collection( collection_name=coll, ) - print("Searching...") + logger.debug("Searching...") res = self.client.search( collection_name=coll, @@ -128,7 +131,7 @@ class EntityVectors: # If reload time has passed, unload collection if time.time() > self.next_reload: - print("Unloading, reload at", self.next_reload) + logger.debug(f"Unloading, reload at {self.next_reload}") self.client.release_collection( collection_name=coll, ) diff --git a/trustgraph-flow/trustgraph/direct/milvus_object_embeddings.py b/trustgraph-flow/trustgraph/direct/milvus_object_embeddings.py index 92cacfc7..290f5155 100644 --- a/trustgraph-flow/trustgraph/direct/milvus_object_embeddings.py +++ b/trustgraph-flow/trustgraph/direct/milvus_object_embeddings.py @@ -1,6 +1,9 @@ from pymilvus import MilvusClient, CollectionSchema, FieldSchema, DataType import time +import logging + +logger = logging.getLogger(__name__) class ObjectVectors: @@ -21,7 +24,7 @@ class ObjectVectors: # Next time to reload - this forces a reload at next window self.next_reload = time.time() + self.reload_time - print("Reload at", self.next_reload) + logger.debug(f"Reload at {self.next_reload}") def init_collection(self, dimension, name): @@ -126,12 +129,12 @@ class ObjectVectors: } } - print("Loading...") + logger.debug("Loading...") self.client.load_collection( collection_name=coll, ) - print("Searching...") + logger.debug("Searching...") res = self.client.search( collection_name=coll, @@ -144,7 +147,7 @@ class ObjectVectors: # If reload time has passed, unload collection if time.time() > self.next_reload: - print("Unloading, reload at", self.next_reload) + logger.debug(f"Unloading, reload at {self.next_reload}") self.client.release_collection( collection_name=coll, ) diff --git a/trustgraph-flow/trustgraph/embeddings/document_embeddings/embeddings.py b/trustgraph-flow/trustgraph/embeddings/document_embeddings/embeddings.py index 95e5462d..602f7bb8 100755 --- a/trustgraph-flow/trustgraph/embeddings/document_embeddings/embeddings.py +++ b/trustgraph-flow/trustgraph/embeddings/document_embeddings/embeddings.py @@ -11,6 +11,10 @@ from ... schema import EmbeddingsRequest, EmbeddingsResponse from ... base import FlowProcessor, RequestResponseSpec, ConsumerSpec from ... base import ProducerSpec +import logging + +logger = logging.getLogger(__name__) + default_ident = "document-embeddings" class Processor(FlowProcessor): @@ -52,7 +56,7 @@ class Processor(FlowProcessor): async def on_message(self, msg, consumer, flow): v = msg.value() - print(f"Indexing {v.metadata.id}...", flush=True) + logger.info(f"Indexing {v.metadata.id}...") try: @@ -79,12 +83,12 @@ class Processor(FlowProcessor): await flow("output").send(r) except Exception as e: - print("Exception:", e, flush=True) + logger.error("Exception occurred", exc_info=True) # Retry raise e - print("Done.", flush=True) + logger.info("Done.") @staticmethod def add_args(parser): diff --git a/trustgraph-flow/trustgraph/embeddings/graph_embeddings/embeddings.py b/trustgraph-flow/trustgraph/embeddings/graph_embeddings/embeddings.py index 043be3a7..4726be4d 100755 --- a/trustgraph-flow/trustgraph/embeddings/graph_embeddings/embeddings.py +++ b/trustgraph-flow/trustgraph/embeddings/graph_embeddings/embeddings.py @@ -11,6 +11,10 @@ from ... schema import EmbeddingsRequest, EmbeddingsResponse from ... base import FlowProcessor, EmbeddingsClientSpec, ConsumerSpec from ... base import ProducerSpec +import logging + +logger = logging.getLogger(__name__) + default_ident = "graph-embeddings" class Processor(FlowProcessor): @@ -50,7 +54,7 @@ class Processor(FlowProcessor): async def on_message(self, msg, consumer, flow): v = msg.value() - print(f"Indexing {v.metadata.id}...", flush=True) + logger.info(f"Indexing {v.metadata.id}...") entities = [] @@ -77,12 +81,12 @@ class Processor(FlowProcessor): await flow("output").send(r) except Exception as e: - print("Exception:", e, flush=True) + logger.error("Exception occurred", exc_info=True) # Retry raise e - print("Done.", flush=True) + logger.info("Done.") @staticmethod def add_args(parser): diff --git a/trustgraph-flow/trustgraph/external/wikipedia/service.py b/trustgraph-flow/trustgraph/external/wikipedia/service.py index f7de78da..d2b5b415 100644 --- a/trustgraph-flow/trustgraph/external/wikipedia/service.py +++ b/trustgraph-flow/trustgraph/external/wikipedia/service.py @@ -10,6 +10,9 @@ from trustgraph.schema import encyclopedia_lookup_response_queue from trustgraph.log_level import LogLevel from trustgraph.base import ConsumerProducer import requests +import logging + +logger = logging.getLogger(__name__) module = "wikipedia" @@ -46,7 +49,7 @@ class Processor(ConsumerProducer): # Sender-produced ID id = msg.properties()["id"] - print(f"Handling {v.kind} / {v.term}...", flush=True) + logger.info(f"Handling {v.kind} / {v.term}...") try: diff --git a/trustgraph-flow/trustgraph/processing/processing.py b/trustgraph-flow/trustgraph/processing/processing.py index 5352776a..8ee62cdd 100644 --- a/trustgraph-flow/trustgraph/processing/processing.py +++ b/trustgraph-flow/trustgraph/processing/processing.py @@ -11,9 +11,13 @@ import importlib from .. log_level import LogLevel +import logging + +logger = logging.getLogger(__name__) + def fn(module_name, class_name, params, w): - print(f"Starting {module_name}...") + logger.info(f"Starting {module_name}...") if "log_level" in params: params["log_level"] = LogLevel(params["log_level"]) @@ -22,7 +26,7 @@ def fn(module_name, class_name, params, w): try: - print(f"Starting {class_name} using {module_name}...") + logger.info(f"Starting {class_name} using {module_name}...") module = importlib.import_module(module_name) class_object = getattr(module, class_name) @@ -30,16 +34,16 @@ def fn(module_name, class_name, params, w): processor = class_object(**params) processor.run() - print(f"{module_name} stopped.") + logger.info(f"{module_name} stopped.") except Exception as e: - print("Exception:", e) + logger.error("Exception occurred", exc_info=True) - print("Restarting in 10...") + logger.info("Restarting in 10...") time.sleep(10) - print("Closing") + logger.info("Closing") w.close() class Processing: @@ -108,7 +112,7 @@ class Processing: readers.remove(r) wait_for -= 1 - print("All processes exited") + logger.info("All processes exited") for p in procs: p.join() @@ -169,13 +173,12 @@ def run(): p.run() - print("Finished.") + logger.info("Finished.") break except Exception as e: - print("Exception:", e, flush=True) - print("Will retry...", flush=True) + logger.error("Exception occurred, will retry...", exc_info=True) time.sleep(10) diff --git a/trustgraph-flow/trustgraph/tables/config.py b/trustgraph-flow/trustgraph/tables/config.py index 45dfc4d9..c0c0a84a 100644 --- a/trustgraph-flow/trustgraph/tables/config.py +++ b/trustgraph-flow/trustgraph/tables/config.py @@ -9,6 +9,9 @@ from ssl import SSLContext, PROTOCOL_TLSv1_2 import uuid import time import asyncio +import logging + +logger = logging.getLogger(__name__) class ConfigTableStore: @@ -19,7 +22,7 @@ class ConfigTableStore: self.keyspace = keyspace - print("Connecting to Cassandra...", flush=True) + logger.info("Connecting to Cassandra...") if cassandra_user and cassandra_password: ssl_context = SSLContext(PROTOCOL_TLSv1_2) @@ -36,7 +39,7 @@ class ConfigTableStore: self.cassandra = self.cluster.connect() - print("Connected.", flush=True) + logger.info("Connected.") self.ensure_cassandra_schema() @@ -44,9 +47,9 @@ class ConfigTableStore: def ensure_cassandra_schema(self): - print("Ensure Cassandra schema...", flush=True) + logger.debug("Ensure Cassandra schema...") - print("Keyspace...", flush=True) + logger.debug("Keyspace...") # FIXME: Replication factor should be configurable self.cassandra.execute(f""" @@ -59,7 +62,7 @@ class ConfigTableStore: self.cassandra.set_keyspace(self.keyspace) - print("config table...", flush=True) + logger.debug("config table...") self.cassandra.execute(""" CREATE TABLE IF NOT EXISTS config ( @@ -70,7 +73,7 @@ class ConfigTableStore: ); """); - print("version table...", flush=True) + logger.debug("version table...") self.cassandra.execute(""" CREATE TABLE IF NOT EXISTS version ( @@ -84,14 +87,14 @@ class ConfigTableStore: SELECT version FROM version """) - print("ensure version...", flush=True) + logger.debug("ensure version...") self.cassandra.execute(""" UPDATE version set version = version + 0 WHERE id = 'version' """) - print("Cassandra schema OK.", flush=True) + logger.info("Cassandra schema OK.") async def inc_version(self): @@ -160,10 +163,8 @@ class ConfigTableStore: except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) async def get_value(self, cls, key): @@ -180,10 +181,8 @@ class ConfigTableStore: except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) for row in resp: return row[0] @@ -205,10 +204,8 @@ class ConfigTableStore: except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) return [ [row[0], row[1]] @@ -230,10 +227,8 @@ class ConfigTableStore: except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) return [ row[0] for row in resp @@ -254,10 +249,8 @@ class ConfigTableStore: except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) return [ (row[0], row[1], row[2]) @@ -279,10 +272,8 @@ class ConfigTableStore: except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) return [ row[0] for row in resp @@ -302,8 +293,6 @@ class ConfigTableStore: break except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) diff --git a/trustgraph-flow/trustgraph/tables/knowledge.py b/trustgraph-flow/trustgraph/tables/knowledge.py index 36414dc4..dc83dbf2 100644 --- a/trustgraph-flow/trustgraph/tables/knowledge.py +++ b/trustgraph-flow/trustgraph/tables/knowledge.py @@ -9,6 +9,9 @@ from ssl import SSLContext, PROTOCOL_TLSv1_2 import uuid import time import asyncio +import logging + +logger = logging.getLogger(__name__) class KnowledgeTableStore: @@ -19,7 +22,7 @@ class KnowledgeTableStore: self.keyspace = keyspace - print("Connecting to Cassandra...", flush=True) + logger.info("Connecting to Cassandra...") if cassandra_user and cassandra_password: ssl_context = SSLContext(PROTOCOL_TLSv1_2) @@ -36,7 +39,7 @@ class KnowledgeTableStore: self.cassandra = self.cluster.connect() - print("Connected.", flush=True) + logger.info("Connected.") self.ensure_cassandra_schema() @@ -44,9 +47,9 @@ class KnowledgeTableStore: def ensure_cassandra_schema(self): - print("Ensure Cassandra schema...", flush=True) + logger.debug("Ensure Cassandra schema...") - print("Keyspace...", flush=True) + logger.debug("Keyspace...") # FIXME: Replication factor should be configurable self.cassandra.execute(f""" @@ -59,7 +62,7 @@ class KnowledgeTableStore: self.cassandra.set_keyspace(self.keyspace) - print("triples table...", flush=True) + logger.debug("triples table...") self.cassandra.execute(""" CREATE TABLE IF NOT EXISTS triples ( @@ -77,7 +80,7 @@ class KnowledgeTableStore: ); """); - print("graph_embeddings table...", flush=True) + logger.debug("graph_embeddings table...") self.cassandra.execute(""" create table if not exists graph_embeddings ( @@ -103,7 +106,7 @@ class KnowledgeTableStore: graph_embeddings ( user ); """); - print("document_embeddings table...", flush=True) + logger.debug("document_embeddings table...") self.cassandra.execute(""" create table if not exists document_embeddings ( @@ -129,7 +132,7 @@ class KnowledgeTableStore: document_embeddings ( user ); """); - print("Cassandra schema OK.", flush=True) + logger.info("Cassandra schema OK.") def prepare_statements(self): @@ -231,10 +234,8 @@ class KnowledgeTableStore: except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) async def add_graph_embeddings(self, m): @@ -276,10 +277,8 @@ class KnowledgeTableStore: except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) async def add_document_embeddings(self, m): @@ -321,14 +320,12 @@ class KnowledgeTableStore: except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) async def list_kg_cores(self, user): - print("List kg cores...") + logger.debug("List kg cores...") while True: @@ -342,10 +339,8 @@ class KnowledgeTableStore: break except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) lst = [ @@ -353,13 +348,13 @@ class KnowledgeTableStore: for row in resp ] - print("Done") + logger.debug("Done") return lst async def delete_kg_core(self, user, document_id): - print("Delete kg cores...") + logger.debug("Delete kg cores...") while True: @@ -373,10 +368,8 @@ class KnowledgeTableStore: break except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) while True: @@ -390,14 +383,12 @@ class KnowledgeTableStore: break except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) async def get_triples(self, user, document_id, receiver): - print("Get triples...") + logger.debug("Get triples...") while True: @@ -411,10 +402,8 @@ class KnowledgeTableStore: break except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) for row in resp: @@ -451,11 +440,11 @@ class KnowledgeTableStore: ) ) - print("Done") + logger.debug("Done") async def get_graph_embeddings(self, user, document_id, receiver): - print("Get GE...") + logger.debug("Get GE...") while True: @@ -469,10 +458,8 @@ class KnowledgeTableStore: break except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) for row in resp: @@ -508,5 +495,5 @@ class KnowledgeTableStore: ) ) - print("Done") + logger.debug("Done") diff --git a/trustgraph-flow/trustgraph/tables/library.py b/trustgraph-flow/trustgraph/tables/library.py index c8cdb027..b186d063 100644 --- a/trustgraph-flow/trustgraph/tables/library.py +++ b/trustgraph-flow/trustgraph/tables/library.py @@ -13,6 +13,9 @@ from ssl import SSLContext, PROTOCOL_TLSv1_2 import uuid import time import asyncio +import logging + +logger = logging.getLogger(__name__) class LibraryTableStore: @@ -23,7 +26,7 @@ class LibraryTableStore: self.keyspace = keyspace - print("Connecting to Cassandra...", flush=True) + logger.info("Connecting to Cassandra...") if cassandra_user and cassandra_password: ssl_context = SSLContext(PROTOCOL_TLSv1_2) @@ -40,7 +43,7 @@ class LibraryTableStore: self.cassandra = self.cluster.connect() - print("Connected.", flush=True) + logger.info("Connected.") self.ensure_cassandra_schema() @@ -48,9 +51,9 @@ class LibraryTableStore: def ensure_cassandra_schema(self): - print("Ensure Cassandra schema...", flush=True) + logger.debug("Ensure Cassandra schema...") - print("Keyspace...", flush=True) + logger.debug("Keyspace...") # FIXME: Replication factor should be configurable self.cassandra.execute(f""" @@ -63,7 +66,7 @@ class LibraryTableStore: self.cassandra.set_keyspace(self.keyspace) - print("document table...", flush=True) + logger.debug("document table...") self.cassandra.execute(""" CREATE TABLE IF NOT EXISTS document ( @@ -82,14 +85,14 @@ class LibraryTableStore: ); """); - print("object index...", flush=True) + logger.debug("object index...") self.cassandra.execute(""" CREATE INDEX IF NOT EXISTS document_object ON document (object_id) """); - print("processing table...", flush=True) + logger.debug("processing table...") self.cassandra.execute(""" CREATE TABLE IF NOT EXISTS processing ( @@ -104,7 +107,7 @@ class LibraryTableStore: ); """); - print("Cassandra schema OK.", flush=True) + logger.info("Cassandra schema OK.") def prepare_statements(self): @@ -204,7 +207,7 @@ class LibraryTableStore: async def add_document(self, document, object_id): - print("Adding document", document.id, object_id) + logger.info(f"Adding document {document.id} {object_id}") metadata = [ ( @@ -231,16 +234,14 @@ class LibraryTableStore: except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) - print("Add complete", flush=True) + logger.debug("Add complete") async def update_document(self, document): - print("Updating document", document.id) + logger.info(f"Updating document {document.id}") metadata = [ ( @@ -267,16 +268,14 @@ class LibraryTableStore: except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) - print("Update complete", flush=True) + logger.debug("Update complete") async def remove_document(self, user, document_id): - print("Removing document", document_id) + logger.info(f"Removing document {document_id}") while True: @@ -293,16 +292,14 @@ class LibraryTableStore: except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) - print("Delete complete", flush=True) + logger.debug("Delete complete") async def list_documents(self, user): - print("List documents...") + logger.debug("List documents...") while True: @@ -316,10 +313,8 @@ class LibraryTableStore: break except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) lst = [ @@ -344,13 +339,13 @@ class LibraryTableStore: for row in resp ] - print("Done") + logger.debug("Done") return lst async def get_document(self, user, id): - print("Get document") + logger.debug("Get document") while True: @@ -364,10 +359,8 @@ class LibraryTableStore: break except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) for row in resp: @@ -390,14 +383,14 @@ class LibraryTableStore: object_id = row[6], ) - print("Done") + logger.debug("Done") return doc raise RuntimeError("No such document row?") async def get_document_object_id(self, user, id): - print("Get document obj ID") + logger.debug("Get document obj ID") while True: @@ -411,14 +404,12 @@ class LibraryTableStore: break except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) for row in resp: - print("Done") + logger.debug("Done") return row[6] raise RuntimeError("No such document row?") @@ -440,7 +431,7 @@ class LibraryTableStore: async def add_processing(self, processing): - print("Adding processing", processing.id) + logger.info(f"Adding processing {processing.id}") while True: @@ -460,16 +451,14 @@ class LibraryTableStore: except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) - print("Add complete", flush=True) + logger.debug("Add complete") async def remove_processing(self, user, processing_id): - print("Removing processing", processing_id) + logger.info(f"Removing processing {processing_id}") while True: @@ -486,16 +475,14 @@ class LibraryTableStore: except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) - print("Delete complete", flush=True) + logger.debug("Delete complete") async def list_processing(self, user): - print("List processing objects") + logger.debug("List processing objects") while True: @@ -509,10 +496,8 @@ class LibraryTableStore: break except Exception as e: - print("Exception:", type(e)) + logger.error("Exception occurred", exc_info=True) raise e - print(f"{e}, retry...", flush=True) - await asyncio.sleep(1) lst = [ @@ -528,7 +513,7 @@ class LibraryTableStore: for row in resp ] - print("Done") + logger.debug("Done") return lst