diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py b/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py index 268c6b04..5e0b30ba 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py @@ -46,9 +46,8 @@ class LibrarianRequestor(ServiceRequestor): criteria = None if "content" in body: - content = base64.b64decode( - body["content"].encode("utf-8") - ) + content = base64.b64decode(body["content"].encode("utf-8")) + content = base64.b64encode(content).decode("utf-8"), else: content = None @@ -58,7 +57,7 @@ class LibrarianRequestor(ServiceRequestor): processing_id = body.get("processing-id", None), document_metadata = dm, processing_metadata = pm, - content = base64.b64encode(content).decode("utf-8"), + content = content, user = body.get("user", None), collection = body.get("collection", None), criteria = criteria, diff --git a/trustgraph-flow/trustgraph/librarian/librarian.py b/trustgraph-flow/trustgraph/librarian/librarian.py index c498a37a..026a34c5 100644 --- a/trustgraph-flow/trustgraph/librarian/librarian.py +++ b/trustgraph-flow/trustgraph/librarian/librarian.py @@ -51,11 +51,6 @@ class Librarian: self.table_store.add_document(request.document_metadata, object_id) - # 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( @@ -70,7 +65,28 @@ class Librarian: raise RuntimeError("Not implemented") async def update_document(self, request): - raise RuntimeError("Not implemented") + + print("UPDATING...") + + # You can't update the document ID, user or kind. + + if not 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) + + print("Update complete", flush=True) + + return LibrarianResponse( + error = None, + document_metadata = None, + content = None, + document_metadatas = None, + processing_metadatas = None, + ) async def get_document_metadata(self, request): raise RuntimeError("Not implemented") @@ -81,6 +97,11 @@ class Librarian: async def add_processing(self, request): raise RuntimeError("Not implemented") + # if document.kind == "application/pdf": + # await self.load_document(document) + # elif document.kind == "text/plain": + # await self.load_text(document) + async def remove_processing(self, request): raise RuntimeError("Not implemented") diff --git a/trustgraph-flow/trustgraph/librarian/service.py b/trustgraph-flow/trustgraph/librarian/service.py index d0d30552..d077dba4 100755 --- a/trustgraph-flow/trustgraph/librarian/service.py +++ b/trustgraph-flow/trustgraph/librarian/service.py @@ -207,6 +207,7 @@ class Processor(AsyncProcessor): raise RequestError(f"Invalid operation: {v.operation}") print("HANDLING...") + return await impls[v.operation](v) async def on_librarian_request(self, msg, consumer, flow): diff --git a/trustgraph-flow/trustgraph/librarian/table_store.py b/trustgraph-flow/trustgraph/librarian/table_store.py index f59e09f9..4ced4d65 100644 --- a/trustgraph-flow/trustgraph/librarian/table_store.py +++ b/trustgraph-flow/trustgraph/librarian/table_store.py @@ -179,6 +179,13 @@ class TableStore: VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """) + self.update_document_stmt = self.cassandra.prepare(""" + UPDATE document + SET time = ?, title = ?, comments = ?, + metadata = ?, tags = ? + WHERE user = ? AND id = ? + """) + self.get_document_stmt = self.cassandra.prepare(""" SELECT user, time, kind, title, comments, metadata, tags, object_id @@ -254,9 +261,6 @@ class TableStore: def add_document(self, document, object_id): - # Create timestamp - when = int(time.time() * 1000) - print("Adding document", document.id, object_id) metadata = [ @@ -274,7 +278,7 @@ class TableStore: resp = self.cassandra.execute( self.insert_document_stmt, ( - document.id, document.user, when, + document.id, document.user, int(document.time * 1000), document.kind, document.title, document.comments, metadata, document.tags, object_id ) @@ -290,6 +294,41 @@ class TableStore: print("Add complete", flush=True) + def update_document(self, document): + + print("Updating document", document.id) + + metadata = [ + ( + v.s.value, v.s.is_uri, v.p.value, v.p.is_uri, + v.o.value, v.o.is_uri + ) + for v in document.metadata + ] + + while True: + + try: + + resp = self.cassandra.execute( + self.update_document_stmt, + ( + int(document.time * 1000), document.title, + document.comments, metadata, document.tags, + document.user, document.id + ) + ) + + break + + except Exception as e: + + print("Exception:", type(e)) + print(f"{e}, retry...", flush=True) + time.sleep(1) + + print("Update complete", flush=True) + def add_triples(self, m): when = int(time.time() * 1000)