diff --git a/test-api/test-library-list b/test-api/test-library-list index 72ea4478..cecb835f 100755 --- a/test-api/test-library-list +++ b/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, } diff --git a/trustgraph-base/trustgraph/schema/library.py b/trustgraph-base/trustgraph/schema/library.py index 91105816..36a4701a 100644 --- a/trustgraph-base/trustgraph/schema/library.py +++ b/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() diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py b/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py index 0a9f314b..3be6aca3 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/librarian.py +++ b/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, diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/serialize.py b/trustgraph-flow/trustgraph/gateway/dispatch/serialize.py index 7c93e75e..8f1eba41 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/serialize.py +++ b/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), diff --git a/trustgraph-flow/trustgraph/librarian/service.py b/trustgraph-flow/trustgraph/librarian/service.py index 73ae9e5a..80334edd 100755 --- a/trustgraph-flow/trustgraph/librarian/service.py +++ b/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: diff --git a/trustgraph-flow/trustgraph/librarian/table_store.py b/trustgraph-flow/trustgraph/librarian/table_store.py index 9da2613c..ed1fc17b 100644 --- a/trustgraph-flow/trustgraph/librarian/table_store.py +++ b/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>, - PRIMARY KEY (user, collection, id) + tags list, + 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 (