From 4b63806c4ad8521f471246a4acd8163846d63ffa Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Sat, 3 May 2025 20:15:04 +0100 Subject: [PATCH] Implementing new schema --- trustgraph-base/trustgraph/schema/library.py | 20 ++-- .../trustgraph/gateway/dispatch/librarian.py | 51 +++++++--- .../trustgraph/gateway/dispatch/serialize.py | 98 ++++++++----------- .../trustgraph/librarian/librarian.py | 37 ++++++- .../trustgraph/librarian/service.py | 42 +++----- .../trustgraph/librarian/table_store.py | 4 +- 6 files changed, 136 insertions(+), 116 deletions(-) diff --git a/trustgraph-base/trustgraph/schema/library.py b/trustgraph-base/trustgraph/schema/library.py index fd0c6289..91105816 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() @@ -77,17 +77,7 @@ class ProcessingMetadata(Record): flow = String() user = String() collection = String() - -class DocumentInfo(Record): - id = String() - flow = String() - kind = String() - user = String() - collection = String() - title = String() - comments = String() - time = Long() - metadata = Array(Triple()) + tags = Array(String()) class Criteria(Record): key = String() @@ -128,8 +118,10 @@ class LibrarianRequest(Record): class LibrarianResponse(Record): error = Error() - document = DocumentPackage() - info = Array(DocumentInfo()) + document_metadata = DocumentMetadata() + content = Bytes() + document_metadatas = Array(DocumentMetadata()) + processing_metadatas = Array(ProcessingMetadata()) librarian_request_queue = topic( 'librarian', kind='non-persistent', namespace='request' diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py b/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py index 5f19ee42..0a9f314b 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py @@ -23,21 +23,35 @@ class LibrarianRequestor(ServiceRequestor): def to_request(self, body): - if "document" in body: - dp = to_document_package(body["document"]) + if "document-metadata" in body: + dm = to_document_metadata(body["document-metadata"]) else: - dp = None + dm = None + + if "processing-metadata" in body: + pm = to_processing_metadata(body["processing-metadata"]) + else: + pm = None if "criteria" in body: criteria = to_criteria(body["criteria"]) else: criteria = None + if "content" in body: + content = base64.b64decode( + body["content"].decode("utf-8") + ).encode("utf-8") + else: + content = None + return LibrarianRequest( operation = body.get("operation", None), - id = body.get("id", None), - flow = body.get("flow", None), - document = dp, + document_id = body.get("document-id", None), + processing_id = body.get("processing-id", None), + document_metadata = dm, + processing_metadata = pm, + content = content user = body.get("user", None), collection = body.get("collection", None), criteria = criteria, @@ -47,13 +61,26 @@ class LibrarianRequestor(ServiceRequestor): response = {} - if message.document: - response["document"] = serialize_document_package(message.document) + if message.document_metadata: + response["document-metadata"] = serialize_document_metadata( + message.document_metadata + ) - if message.info: - response["info"] = [ - serialize_document_info(v) - for v in message.info + if message.content: + response["content"] = base64.b64encode( + message["content"].decode("utf-8") + ).encode("utf-8") + + if message.document_metadatas: + response["document-metadatas"] = [ + serialize_document_metadata(v) + for v in message.document_metadatas + ] + + if message.processing_metadatas: + response["processing-metadatas"] = [ + serialize_processing_metadata(v) + for v in message.processing_metadatas ] return response, True diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/serialize.py b/trustgraph-flow/trustgraph/gateway/dispatch/serialize.py index 799bb87e..7c93e75e 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/serialize.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/serialize.py @@ -80,96 +80,78 @@ def serialize_document_embeddings(message): ], } -def serialize_document_package(message): +def serialize_document_metadata(message): ret = {} - if message.id: - ret["id"] = message.id +# if message.id: +# ret["id"] = message.id - if message.metadata: - ret["metadata"] = serialize_subgraph(message.metdata) - - if message.document: - blob = base64.b64encode( - message.document.encode("utf-8") - ).decode("utf-8") - ret["document"] = blob + if message.time: + ret["time"] = message.time if message.kind: ret["kind"] = message.kind - if message.flow: - ret["flow"] = message.flow - - if message.user: - ret["user"] = message.user - - if message.collection: - ret["collection"] = message.collection - - return ret - -def serialize_document_info(message): - - ret = {} - - if message.id: - ret["id"] = message.id - - if message.kind: - ret["kind"] = message.kind - - if message.flow: - ret["flow"] = message.flow - - if message.user: - ret["user"] = message.user - - if message.collection: - ret["collection"] = message.collection - if message.title: ret["title"] = message.title if message.comments: ret["comments"] = message.comments - if message.time: - ret["time"] = message.time - if message.metadata: ret["metadata"] = serialize_subgraph(message.metadata) + if message.tags: + ret["tags"] = message.tags + return ret -def to_document_package(x): +def serialize_processing_metadata(message): + + ret = {} + + if message.id: + ret["document_id"] = message.document_id + + if message.time: + ret["time"] = message.time + + if message.kind: + ret["flow"] = message.flow + + if message.user: + ret["user"] = message.user + + if message.collection: + ret["collection"] = message.collection + + if message.tags: + ret["tags"] = message.tags + + return ret + +def to_document_metadata(x): return DocumentPackage( - id = x.get("id", None), - flow = x.get("flow", None), +# id = x.get("id", None), + time = x.get("time", None), kind = x.get("kind", None), - user = x.get("user", None), - collection = x.get("collection", None), title = x.get("title", None), comments = x.get("comments", None), - time = x.get("time", None), - document = x.get("document", None), metadata = to_subgraph(x["metadata"]), + tags = x.get("tags", None), ) -def to_document_info(x): +def to_processing_metadata(x): return DocumentInfo( - id = x.get("id", None), + document_id = x.get("document_id", None), + time = x.get("time", None), flow = x.get("flow", None), - kind = x.get("kind", None), user = x.get("user", None), collection = x.get("collection", None), - title = x.get("title", None), - comments = x.get("comments", None), - time = x.get("time", None), - metadata = to_subgraph(x["metadata"]), + tags = x.get("tags", None), ) def to_criteria(x): diff --git a/trustgraph-flow/trustgraph/librarian/librarian.py b/trustgraph-flow/trustgraph/librarian/librarian.py index 7a22605c..9475d26d 100644 --- a/trustgraph-flow/trustgraph/librarian/librarian.py +++ b/trustgraph-flow/trustgraph/librarian/librarian.py @@ -26,17 +26,19 @@ class Librarian: self.load_document = load_document self.load_text = load_text - async def add(self, document): + async def add_document(self, request): + raise RuntimeError("Not implemented") if document.kind not in ( "text/plain", "application/pdf" ): raise RequestError("Invalid document kind: " + document.kind) - # Create object ID as a hash of the document - object_id = uuid.UUID(hash(document.document)) + # Create object ID for blob + object_id = str(uuid.uuid4()) - self.blob_store.add(object_id, document.document, document.kind) + self.blob_store.add(object_id, request.content, + request.document_metadata.kind) self.table_store.add(object_id, document) @@ -52,9 +54,36 @@ class Librarian: document = None, info = None, ) + async def remove_document(self, request): + raise RuntimeError("Not implemented") + + async def update_document(self, request): + raise RuntimeError("Not implemented") + + async def get_document_metadata(self, request): + raise RuntimeError("Not implemented") + + async def get_document_content(self, request): + raise RuntimeError("Not implemented") + + async def add_processing(self, request): + raise RuntimeError("Not implemented") + + async def remove_processing(self, request): + raise RuntimeError("Not implemented") + + async def list_documents(self, request): + raise RuntimeError("Not implemented") + + async def list_processing(self, request): + raise RuntimeError("Not implemented") async def list(self, user, collection): + raise RuntimeError("Not implemented") + + + print("list") info = self.table_store.list(user, collection) diff --git a/trustgraph-flow/trustgraph/librarian/service.py b/trustgraph-flow/trustgraph/librarian/service.py index 698a32fe..73ae9e5a 100755 --- a/trustgraph-flow/trustgraph/librarian/service.py +++ b/trustgraph-flow/trustgraph/librarian/service.py @@ -184,39 +184,29 @@ class Processor(AsyncProcessor): self.text_load.send(None, doc) - def parse_request(self, v): + def process_request(self, v): if v.operation is None: raise RequestError("Null operation") print("op", v.operation) - if v.operation == "add": - if ( - v.document and v.document.id and v.document.metadata and - v.document.document and v.document.kind - ): - return partial( - self.librarian.add, - document = v.document, - ) - else: - raise RequestError("Invalid call") + impls = { + "add-document": self.librarian.add_document, + "remove-document": self.librarian.remove_document, + "update-document": self.librarian.update_document, + "get-document-metadata": self.librarian.get_document_metadata, + "get-document-content": self.librarian.get_document_content, + "add-processing": self.librarian.add_processing, + "remove-processing": self.librarian.remove_processing, + "list-documents": self.librarian.list_documents, + "list-processing": self.librarian.list_processing, + } - if v.operation == "list": - print("list", v) - print(v.user) - if v.user: - return partial( - self.librarian.list, - user = v.user, - collection = v.collection, - ) - else: - print("User not specified") - raise RequestError("Invalid call") + if v.operation not in impls: + raise RequestError(f"Invalid operation: {v.operation}") - raise RequestError("Invalid operation: " + v.operation) + return impls[v.operation](v) async def on_librarian_request(self, msg, consumer, flow): @@ -231,7 +221,7 @@ class Processor(AsyncProcessor): print(f"Handling input {id}...", flush=True) try: - func = self.parse_request(v) + func = self.process_request(v) except RequestError as e: resp = LibrarianResponse( error = Error( diff --git a/trustgraph-flow/trustgraph/librarian/table_store.py b/trustgraph-flow/trustgraph/librarian/table_store.py index 823d5cc0..9da2613c 100644 --- a/trustgraph-flow/trustgraph/librarian/table_store.py +++ b/trustgraph-flow/trustgraph/librarian/table_store.py @@ -1,5 +1,5 @@ from .. schema import LibrarianRequest, LibrarianResponse -from .. schema import DocumentInfo, Error, Triple, Value +from .. schema import DocumentMetadata, Error, Triple, Value from .. knowledge import hash from .. exceptions import RequestError @@ -325,7 +325,7 @@ class TableStore: print("OK2") info = [ - DocumentInfo( + DocumentMetadata( id = row[0], kind = row[1], flow = row[2],