Tweaking schema

test-api/test-library-list

@@ -12,7 +12,7 @@ url = "http://localhost:8088/api/v1/"
 user = "trustgraph"
 
 input = {
-    "operation": "list",
+    "operation": "list-documents",
     "user": user,
 }
 

trustgraph-base/trustgraph/schema/library.py

@@ -91,6 +91,8 @@ class LibrarianRequest(Record):
     # list-processing
     operation = String()
 
+    user = String()
+
     # add-document, remove-document, update-document, get-document-metadata,
     # get-document-content
     document_id = String()

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

@@ -4,8 +4,10 @@ from ... schema import librarian_request_queue
 from ... schema import librarian_response_queue
 
 from . requestor import ServiceRequestor
-from . serialize import serialize_document_package, serialize_document_info
-from . serialize import to_document_package, to_document_info, to_criteria
+from . serialize import serialize_document_metadata
+from . serialize import serialize_processing_metadata
+from . serialize import to_document_metadata, to_processing_metadata
+from . serialize import to_criteria
 
 class LibrarianRequestor(ServiceRequestor):
     def __init__(self, pulsar_client, consumer, subscriber, timeout=120):
@@ -51,7 +53,7 @@ class LibrarianRequestor(ServiceRequestor):
             processing_id = body.get("processing-id", None),
             document_metadata = dm,
             processing_metadata = pm,
-            content = content
+            content = content,
             user = body.get("user", None),
             collection = body.get("collection", None),
             criteria = criteria,

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

@@ -1,7 +1,7 @@
 
 import base64
 
-from ... schema import Value, Triple, DocumentPackage, DocumentInfo
+from ... schema import Value, Triple, DocumentMetadata, ProcessingMetadata
 
 def to_value(x):
     return Value(value=x["v"], is_uri=x["e"])
@@ -133,7 +133,7 @@ def serialize_processing_metadata(message):
 
 def to_document_metadata(x):
 
-    return DocumentPackage(
+    return DocumentMetadata(
 #        id = x.get("id", None),
         time = x.get("time", None),
         kind = x.get("kind", None),
@@ -145,7 +145,7 @@ def to_document_metadata(x):
 
 def to_processing_metadata(x):
 
-    return DocumentInfo(
+    return ProcessingMetadata(
         document_id = x.get("document_id", None),
         time = x.get("time", None),
         flow = x.get("flow", None),

trustgraph-flow/trustgraph/librarian/service.py

@@ -184,7 +184,7 @@ class Processor(AsyncProcessor):
 
         self.text_load.send(None, doc)
 
-    def process_request(self, v):
+    async def process_request(self, v):
 
         if v.operation is None:
             raise RequestError("Null operation")
@@ -206,7 +206,7 @@ class Processor(AsyncProcessor):
         if v.operation not in impls:
             raise RequestError(f"Invalid operation: {v.operation}")
 
-        return impls[v.operation](v)
+        return await impls[v.operation](v)
 
     async def on_librarian_request(self, msg, consumer, flow):
 
@@ -221,7 +221,7 @@ class Processor(AsyncProcessor):
         print(f"Handling input {id}...", flush=True)
 
         try:
-            func = self.process_request(v)
+            func = await self.process_request(v)
         except RequestError as e:
             resp = LibrarianResponse(
                 error = Error(
@@ -234,6 +234,19 @@ class Processor(AsyncProcessor):
                 resp, properties={"id": id}
             )
 
+            return
+        except Exception as e:
+            resp = LibrarianResponse(
+                error = Error(
+                    type = "unexpected-error",
+                    message = str(e),
+                )
+            )
+
+            await self.librarian_response_producer.send(
+                resp, properties={"id": id}
+            )
+
             return
 
         try:

trustgraph-flow/trustgraph/librarian/table_store.py

@@ -63,19 +63,18 @@ class TableStore:
 
         self.cassandra.execute("""
             CREATE TABLE IF NOT EXISTS document (
-                user text,
-                collection text,
                 id text,
-                flow text,
+                user text,
                 time timestamp,
+                kind text,
                 title text,
                 comments text,
-                kind text,
-                object_id uuid,
                 metadata list<tuple<
                     text, boolean, text, boolean, text, boolean
                 >>,
-                PRIMARY KEY (user, collection, id)
+                tags list<text>,
+                object_id uuid,
+                PRIMARY KEY (user, id)
             );
         """);
 
@@ -86,6 +85,8 @@ class TableStore:
             ON document (object_id)
         """);
 
+        return
+
         print("triples table...", flush=True)
 
         self.cassandra.execute("""
@@ -156,28 +157,29 @@ class TableStore:
         self.insert_document_stmt = self.cassandra.prepare("""
             INSERT INTO document
             (
-                id, user, collection, flow, kind, object_id, time, title,
-                comments, metadata
+                id, user, time, kind, title, comments, metadata, tags,
+                object_id
             )
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
         """)
 
         self.list_document_stmt = self.cassandra.prepare("""
             SELECT
-                id, kind, flow, user, collection, title, comments, time,
-                metadata
+                id, time, kind, title, comments, metadata, tags, object_id
             FROM document
             WHERE user = ?
         """)
 
         self.list_document_by_collection_stmt = self.cassandra.prepare("""
             SELECT
-                id, kind, flow, user, collection, title, comments, time,
-                metadata
+                id, time, kind, title, comments, metadata, tags, object_id
             FROM document
-            WHERE user = ? AND collection = ?
+            WHERE user = ? AND tags CONTAINS ?
+            ALLOW FILTERING
         """)
 
+        return
+
         self.insert_triples_stmt = self.cassandra.prepare("""
             INSERT INTO triples
             (