From ee959e6f1745abd2b9b6279da81ddad390d01e9a Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Sun, 4 May 2025 20:52:16 +0100 Subject: [PATCH] List works --- test-api/test-library-list-documents | 40 ++++++++++ test-api/test-library-update-doc | 75 ++++++++++++++++++ .../trustgraph/gateway/dispatch/librarian.py | 8 +- .../trustgraph/librarian/librarian.py | 34 ++++++-- .../trustgraph/librarian/table_store.py | 78 +++++++------------ 5 files changed, 179 insertions(+), 56 deletions(-) create mode 100755 test-api/test-library-list-documents create mode 100755 test-api/test-library-update-doc diff --git a/test-api/test-library-list-documents b/test-api/test-library-list-documents new file mode 100755 index 00000000..4166856d --- /dev/null +++ b/test-api/test-library-list-documents @@ -0,0 +1,40 @@ +#!/usr/bin/env python3 + +import requests +import json +import sys +import base64 +import time + +url = "http://localhost:8088/api/v1/" + +############################################################################ + +id = "http://trustgraph.ai/doc/9fdee98b-b259-40ac-bcb9-8e82ccedeb04" + +input = { + "operation": "list-documents", + "user": "trustgraph", +} + +resp = requests.post( + f"{url}librarian", + json=input, +) + +print(resp.text) +resp = resp.json() + +print(resp) + +if "error" in resp: + print(f"Error: {resp['error']}") + sys.exit(1) + +# print(resp["response"]) +print(resp) + +sys.exit(0) + +############################################################################ + diff --git a/test-api/test-library-update-doc b/test-api/test-library-update-doc new file mode 100755 index 00000000..eee4170d --- /dev/null +++ b/test-api/test-library-update-doc @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 + +import requests +import json +import sys +import base64 +import time + +url = "http://localhost:8088/api/v1/" + +############################################################################ + +id = "http://trustgraph.ai/doc/9fdee98b-b259-40ac-bcb9-8e82ccedeb04" + +input = { + "operation": "update-document", + "document-metadata": { + "id": id, + "time": int(time.time()), + "title": "Mark's cats - a story", + "comments": "Information about Mark's cats", + "metadata": [ + { + "s": { + "v": id, + "e": True, + }, + "p": { + "v": "http://www.w3.org/2000/01/rdf-schema#label", + "e": True, + }, + "o": { + "v": "Mark's pets", "e": False, + }, + }, + { + "s": { + "v": id, + "e": True, + }, + "p": { + "v": 'https://schema.org/keywords', + "e": True, + }, + "o": { + "v": "cats", "e": False, + }, + }, + ], + "user": "trustgraph", + "tags": ["mark", "cats", "pets"], + }, +} + +resp = requests.post( + f"{url}librarian", + json=input, +) + +print(resp.text) +resp = resp.json() + +print(resp) + +if "error" in resp: + print(f"Error: {resp['error']}") + sys.exit(1) + +# print(resp["response"]) +print(resp) + +sys.exit(0) + +############################################################################ + diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py b/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py index 5e0b30ba..18db0896 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py @@ -47,7 +47,7 @@ class LibrarianRequestor(ServiceRequestor): if "content" in body: content = base64.b64decode(body["content"].encode("utf-8")) - content = base64.b64encode(content).decode("utf-8"), + content = base64.b64encode(content).decode("utf-8") else: content = None @@ -65,6 +65,8 @@ class LibrarianRequestor(ServiceRequestor): def from_response(self, message): + print(message) + response = {} if message.document_metadata: @@ -75,13 +77,13 @@ class LibrarianRequestor(ServiceRequestor): if message.content: response["content"] = message.content - if message.document_metadatas: + if message.document_metadatas != None: response["document-metadatas"] = [ serialize_document_metadata(v) for v in message.document_metadatas ] - if message.processing_metadatas: + if message.processing_metadatas != None: response["processing-metadatas"] = [ serialize_processing_metadata(v) for v in message.processing_metadatas diff --git a/trustgraph-flow/trustgraph/librarian/librarian.py b/trustgraph-flow/trustgraph/librarian/librarian.py index 026a34c5..88981f89 100644 --- a/trustgraph-flow/trustgraph/librarian/librarian.py +++ b/trustgraph-flow/trustgraph/librarian/librarian.py @@ -37,7 +37,8 @@ class Librarian: "Invalid document kind: " + request.document_metadata.kind ) - if self.table_store.document_exists( + print("Existence test...") + if await self.table_store.document_exists( request.document_metadata.user, request.document_metadata.id ): @@ -46,10 +47,22 @@ class Librarian: # Create object ID for blob object_id = uuid.uuid4() + print("OID", object_id) + print(request.content) + print("CONT", base64.b64decode(request.content)) + + + print("Add blob...") + + self.blob_store.add(object_id, base64.b64decode(request.content), request.document_metadata.kind) - self.table_store.add_document(request.document_metadata, object_id) + print("Add table...") + + await self.table_store.add_document( + request.document_metadata, object_id + ) print("Add complete", flush=True) @@ -70,13 +83,13 @@ class Librarian: # You can't update the document ID, user or kind. - if not self.table_store.document_exists( + if not await self.table_store.document_exists( request.document_metadata.user, request.document_metadata.id ): raise RuntimeError("Document does not exist") - self.table_store.update_document(request.document_metadata) + await self.table_store.update_document(request.document_metadata) print("Update complete", flush=True) @@ -106,7 +119,18 @@ class Librarian: raise RuntimeError("Not implemented") async def list_documents(self, request): - raise RuntimeError("Not implemented") + + docs = await self.table_store.list_documents(request.user) + + print(docs) + + return LibrarianResponse( + error = None, + document_metadata = None, + content = None, + document_metadatas = docs, + processing_metadatas = None, + ) async def list_processing(self, request): raise RuntimeError("Not implemented") diff --git a/trustgraph-flow/trustgraph/librarian/table_store.py b/trustgraph-flow/trustgraph/librarian/table_store.py index 4ced4d65..73cb3afe 100644 --- a/trustgraph-flow/trustgraph/librarian/table_store.py +++ b/trustgraph-flow/trustgraph/librarian/table_store.py @@ -244,7 +244,7 @@ class TableStore: VALUES (?, ?, ?, ?, ?, ?, ?) """) - def document_exists(self, user, id): + async def document_exists(self, user, id): resp = self.cassandra.execute( self.test_document_exists_stmt, @@ -259,7 +259,7 @@ class TableStore: return False - def add_document(self, document, object_id): + async def add_document(self, document, object_id): print("Adding document", document.id, object_id) @@ -290,11 +290,11 @@ class TableStore: print("Exception:", type(e)) print(f"{e}, retry...", flush=True) - time.sleep(1) + await asyncio.sleep(1) print("Add complete", flush=True) - def update_document(self, document): + async def update_document(self, document): print("Updating document", document.id) @@ -325,11 +325,11 @@ class TableStore: print("Exception:", type(e)) print(f"{e}, retry...", flush=True) - time.sleep(1) + await asyncio.sleep(1) print("Update complete", flush=True) - def add_triples(self, m): + async def add_triples(self, m): when = int(time.time() * 1000) @@ -371,77 +371,59 @@ class TableStore: print("Exception:", type(e)) print(f"{e}, retry...", flush=True) - time.sleep(1) + await asyncio.sleep(1) - def list(self, user, collection=None): + async def list_documents(self, user): print("LIST") + while True: - print("TRY") - - print(self.list_document_stmt) try: - if collection: - resp = self.cassandra.execute( - self.list_document_by_collection_stmt, - (user, collection) - ) - else: - resp = self.cassandra.execute( - self.list_document_stmt, - (user,) - ) - break + resp = self.cassandra.execute( + self.list_document_stmt, + (user,) + ) print("OK") + break except Exception as e: print("Exception:", type(e)) print(f"{e}, retry...", flush=True) - time.sleep(1) + await asyncio.sleep(1) - print("OK2") - info = [ + lst = [ DocumentMetadata( id = row[0], - kind = row[1], - flow = row[2], - user = row[3], - collection = row[4], - title = row[5], - comments = row[6], - time = int(1000 * row[7].timestamp()), + user = user, + time = int(time.mktime(row[1].timetuple())), + kind = row[2], + title = row[3], + comments = row[4], metadata = [ Triple( s=Value(value=m[0], is_uri=m[1]), p=Value(value=m[2], is_uri=m[3]), o=Value(value=m[4], is_uri=m[5]) ) - for m in row[8] + for m in row[5] ], + tags = row[6], + object_id = row[7], ) for row in resp ] print("OK3") - print(info) + print(lst) - # print(info[0].user) - # print(info[0].time) - # print(info[0].kind) - # print(info[0].collection) - # print(info[0].title) - # print(info[0].comments) - # print(info[0].metadata) - # print(info[0].metadata) + return lst - return info - - def add_graph_embeddings(self, m): + async def add_graph_embeddings(self, m): when = int(time.time() * 1000) @@ -483,9 +465,9 @@ class TableStore: print("Exception:", type(e)) print(f"{e}, retry...", flush=True) - time.sleep(1) + await asyncio.sleep(1) - def add_document_embeddings(self, m): + async def add_document_embeddings(self, m): when = int(time.time() * 1000) @@ -527,6 +509,6 @@ class TableStore: print("Exception:", type(e)) print(f"{e}, retry...", flush=True) - time.sleep(1) + await asyncio.sleep(1)