From 385b02021f8f62d099d9b155516f4111b394beda Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Sun, 4 May 2025 19:17:46 +0100 Subject: [PATCH] Working on librarian flow --- test-api/test-library-add-doc | 22 +++++---- test-api/test-library-add-doc2 | 18 ++++--- trustgraph-base/trustgraph/schema/library.py | 3 +- .../trustgraph/gateway/dispatch/librarian.py | 20 ++++++-- .../trustgraph/gateway/dispatch/serialize.py | 10 ++-- .../trustgraph/librarian/librarian.py | 31 +++++++----- .../trustgraph/librarian/service.py | 48 ++++--------------- .../trustgraph/librarian/table_store.py | 41 +++++++++++----- 8 files changed, 106 insertions(+), 87 deletions(-) diff --git a/test-api/test-library-add-doc b/test-api/test-library-add-doc index bd927367..3dc65b83 100755 --- a/test-api/test-library-add-doc +++ b/test-api/test-library-add-doc @@ -4,6 +4,7 @@ import requests import json import sys import base64 +import time url = "http://localhost:8088/api/v1/" @@ -11,13 +12,17 @@ url = "http://localhost:8088/api/v1/" id = "http://trustgraph.ai/doc/12345678" -with open("docs/README.cats") as f: - doc = base64.b64encode(f.read().encode("utf-8")).decode("utf-8") +with open("docs/README.cats", "rb") as f: + doc = base64.b64encode(f.read()).decode("utf-8") input = { - "operation": "add", - "document": { + "operation": "add-document", + "document-metadata": { "id": id, + "time": int(time.time()), + "kind": "text/plain", + "title": "Mark's cats", + "comments": "Test doc taken from the TrustGraph repo", "metadata": [ { "s": { @@ -46,13 +51,10 @@ input = { }, }, ], - "document": doc, - "kind": "text/plain", "user": "trustgraph", - "collection": "default", - "title": "Mark's cats", - "comments": "Test doc taken from the TrustGraph repo", - } + "tags": ["mark", "cats"], + }, + "content": doc, } resp = requests.post( diff --git a/test-api/test-library-add-doc2 b/test-api/test-library-add-doc2 index 0c0856f9..f1761aa4 100755 --- a/test-api/test-library-add-doc2 +++ b/test-api/test-library-add-doc2 @@ -4,6 +4,7 @@ import requests import json import sys import base64 +import time url = "http://localhost:8088/api/v1/" @@ -17,9 +18,13 @@ with open(source, "rb") as f: doc = base64.b64encode(f.read()).decode("utf-8") input = { - "operation": "add", - "id": id, - "document": { + "operation": "add-document", + "document-metadata": { + "id": id, + "time": int(time.time()), + "kind": "application/pdf", + "title": "Application of SAE ARP4754A to Flight Critical Systems", + "comments": "Application of federal safety standards to NASA spacecraft", "metadata": [ { "s": { @@ -61,11 +66,10 @@ input = { }, }, ], - "document": doc, - "kind": "application/pdf", "user": "trustgraph", - "collection": "default", - } + "tags": ["nasa", "safety-engineering"], + }, + "content": doc, } resp = requests.post( diff --git a/trustgraph-base/trustgraph/schema/library.py b/trustgraph-base/trustgraph/schema/library.py index 63900c3a..dfec533f 100644 --- a/trustgraph-base/trustgraph/schema/library.py +++ b/trustgraph-base/trustgraph/schema/library.py @@ -63,7 +63,7 @@ from . documents import Document, TextDocument # search([]) : id[] class DocumentMetadata(Record): -# id = String() + id = String() time = Long() kind = String() title = String() @@ -73,6 +73,7 @@ class DocumentMetadata(Record): tags = Array(String()) class ProcessingMetadata(Record): + id = String() document_id = String() time = Long() flow = String() diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py b/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py index 3be6aca3..9f3799db 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py @@ -1,4 +1,6 @@ +import base64 + from ... schema import LibrarianRequest, LibrarianResponse from ... schema import librarian_request_queue from ... schema import librarian_response_queue @@ -25,6 +27,7 @@ class LibrarianRequestor(ServiceRequestor): def to_request(self, body): + print("to_request") if "document-metadata" in body: dm = to_document_metadata(body["document-metadata"]) else: @@ -40,13 +43,20 @@ class LibrarianRequestor(ServiceRequestor): else: criteria = None + print("a") + print(type(body["content"])) + if "content" in body: content = base64.b64decode( - body["content"].decode("utf-8") - ).encode("utf-8") + body["content"].encode("utf-8") + ) else: content = None + print("b") + + print(type(content)) + return LibrarianRequest( operation = body.get("operation", None), document_id = body.get("document-id", None), @@ -61,6 +71,8 @@ class LibrarianRequestor(ServiceRequestor): def from_response(self, message): + print("from_response") + response = {} if message.document_metadata: @@ -70,8 +82,8 @@ class LibrarianRequestor(ServiceRequestor): if message.content: response["content"] = base64.b64encode( - message["content"].decode("utf-8") - ).encode("utf-8") + message["content"] + ).decode("utf-8") if message.document_metadatas: response["document-metadatas"] = [ diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/serialize.py b/trustgraph-flow/trustgraph/gateway/dispatch/serialize.py index cc869dec..6d7ada54 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/serialize.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/serialize.py @@ -84,8 +84,8 @@ def serialize_document_metadata(message): ret = {} -# if message.id: -# ret["id"] = message.id + if message.id: + ret["id"] = message.id if message.time: ret["time"] = message.time @@ -114,6 +114,9 @@ def serialize_processing_metadata(message): ret = {} + if message.id: + ret["id"] = message.id + if message.id: ret["document_id"] = message.document_id @@ -137,7 +140,7 @@ def serialize_processing_metadata(message): def to_document_metadata(x): return DocumentMetadata( -# id = x.get("id", None), + id = x.get("id", None), time = x.get("time", None), kind = x.get("kind", None), title = x.get("title", None), @@ -150,6 +153,7 @@ def to_document_metadata(x): def to_processing_metadata(x): return ProcessingMetadata( + id = x.get("id", None), document_id = x.get("document_id", None), time = x.get("time", None), flow = x.get("flow", None), diff --git a/trustgraph-flow/trustgraph/librarian/librarian.py b/trustgraph-flow/trustgraph/librarian/librarian.py index 9475d26d..93718191 100644 --- a/trustgraph-flow/trustgraph/librarian/librarian.py +++ b/trustgraph-flow/trustgraph/librarian/librarian.py @@ -1,3 +1,4 @@ + from .. schema import LibrarianRequest, LibrarianResponse, Error, Triple from .. knowledge import hash from .. exceptions import RequestError @@ -27,33 +28,41 @@ class Librarian: self.load_text = load_text async def add_document(self, request): - raise RuntimeError("Not implemented") - if document.kind not in ( + print(request) + + if request.document_metadata.kind not in ( "text/plain", "application/pdf" ): - raise RequestError("Invalid document kind: " + document.kind) + raise RequestError( + "Invalid document kind: " + request.document_metadata.kind + ) # Create object ID for blob - object_id = str(uuid.uuid4()) + object_id = uuid.uuid4() + print("HERE") self.blob_store.add(object_id, request.content, request.document_metadata.kind) - self.table_store.add(object_id, document) + self.table_store.add_document(request.document_metadata, object_id) + print("HERE2") - if document.kind == "application/pdf": - await self.load_document(document) - elif document.kind == "text/plain": - await self.load_text(document) + # if document.kind == "application/pdf": + # await self.load_document(document) + # elif document.kind == "text/plain": + # await self.load_text(document) print("Add complete", flush=True) return LibrarianResponse( error = None, - document = None, - info = None, + document_metadata = None, + content = None, + document_metadatas = None, + processing_metadatas = None, ) + async def remove_document(self, request): raise RuntimeError("Not implemented") diff --git a/trustgraph-flow/trustgraph/librarian/service.py b/trustgraph-flow/trustgraph/librarian/service.py index 80334edd..d0d30552 100755 --- a/trustgraph-flow/trustgraph/librarian/service.py +++ b/trustgraph-flow/trustgraph/librarian/service.py @@ -206,6 +206,7 @@ class Processor(AsyncProcessor): if v.operation not in impls: raise RequestError(f"Invalid operation: {v.operation}") + print("HANDLING...") return await impls[v.operation](v) async def on_librarian_request(self, msg, consumer, flow): @@ -221,7 +222,15 @@ class Processor(AsyncProcessor): print(f"Handling input {id}...", flush=True) try: - func = await self.process_request(v) + + resp = await self.process_request(v) + + await self.librarian_response_producer.send( + resp, properties={"id": id} + ) + + return + except RequestError as e: resp = LibrarianResponse( error = Error( @@ -249,43 +258,6 @@ class Processor(AsyncProcessor): return - try: - resp = await func() - print("->", resp) - except RequestError as e: - resp = LibrarianResponse( - error = Error( - type = "request-error", - message = str(e), - ) - ) - - await self.librarian_response_producer.send( - resp, properties={"id": id} - ) - - return - except Exception as e: - print("Exception:", e, flush=True) - resp = LibrarianResponse( - error = Error( - type = "processing-error", - message = "Unhandled error: " + str(e), - ) - ) - - await self.librarian_response_producer.send( - resp, properties={"id": id} - ) - - return - - print("Send response..!.", flush=True) - - await self.librarian_response_producer.send( - resp, properties={"id": id} - ) - print("Done.", flush=True) @staticmethod diff --git a/trustgraph-flow/trustgraph/librarian/table_store.py b/trustgraph-flow/trustgraph/librarian/table_store.py index ed1fc17b..602f7715 100644 --- a/trustgraph-flow/trustgraph/librarian/table_store.py +++ b/trustgraph-flow/trustgraph/librarian/table_store.py @@ -85,6 +85,21 @@ class TableStore: ON document (object_id) """); + print("processing table...", flush=True) + + self.cassandra.execute(""" + CREATE TABLE IF NOT EXISTS processing ( + id text, + document_id text, + time timestamp, + flow text, + user text, + collection text, + tags list, + PRIMARY KEY (user, id) + ); + """); + return print("triples table...", flush=True) @@ -157,8 +172,9 @@ class TableStore: self.insert_document_stmt = self.cassandra.prepare(""" INSERT INTO document ( - id, user, time, kind, title, comments, metadata, tags, - object_id + id, user, time, + kind, title, comments, + metadata, tags, object_id ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """) @@ -207,17 +223,17 @@ class TableStore: VALUES (?, ?, ?, ?, ?, ?, ?) """) - def add(self, object_id, document): + def add_document(self, document, object_id): - if document.kind not in ( - "text/plain", "application/pdf" - ): - raise RequestError("Invalid document kind: " + document.kind) + # if document.kind not in ( + # "text/plain", "application/pdf" + # ): + # raise RequestError("Invalid document kind: " + document.kind) - # Create random doc ID + # Create timestamp when = int(time.time() * 1000) - print("Adding", document.id, object_id) + print("Adding document", document.id, object_id) metadata = [ ( @@ -234,10 +250,9 @@ class TableStore: resp = self.cassandra.execute( self.insert_document_stmt, ( - document.id, document.user, document.collection, - document.kind, object_id, when, - document.title, document.comments, - metadata + document.id, document.user, when, + document.kind, document.title, document.comments, + metadata, document.tags, object_id ) )