Document update works

trustgraph-flow/trustgraph/gateway/dispatch/librarian.py

@@ -46,9 +46,8 @@ class LibrarianRequestor(ServiceRequestor):
             criteria = None
 
         if "content" in body:
-            content = base64.b64decode(
+            content = base64.b64decode(body["content"].encode("utf-8"))
-                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,

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")
 

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):

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)