mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-24 12:41:02 +02:00
Logging strategy updates
This commit is contained in:
Cyber MacGeddon
parent
1d9aa652e0
commit
05a5127b64
11 changed files with 148 additions and 160 deletions
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue