Knowledge list and fetch working

trustgraph-flow/trustgraph/cores/knowledge.py

@@ -20,7 +20,7 @@ class KnowledgeManager:
 
     async def delete_kg_core(self, request):
 
-        print("Updating doc...")
+        print("Updating doc...", flush=True)
 
         # You can't update the document ID, user or kind.
 
@@ -42,50 +42,69 @@ class KnowledgeManager:
             processing_metadatas = None,
         )
 
-    async def fetch_kg_core(self, request, flow):
+    async def fetch_kg_core(self, request, respond):
 
-        print("Fetch core...")
+        print("Fetch core...", flush=True)
 
-        async def publish(obj):
+        async def publish_triples(t):
-            print(obj)
+            await respond(
-#            await publisher.send(triples)
+                KnowledgeResponse(
+                    error = None,
+                    ids = None,
+                    eos = False,
+                    triples = t,
+                    graph_embeddings = None,
+                )
+            )
 
         # Remove doc table row
         await self.table_store.get_triples(
             request.user,
             request.id,
-            publish
+            publish_triples,
         )
 
-        async def publish_ge(obj):
+        async def publish_ge(g):
-            print(obj)
+            await respond(
+                KnowledgeResponse(
+                    error = None,
+                    ids = None,
+                    eos = False,
+                    triples = None,
+                    graph_embeddings = g,
+                )
+            )
 
         # Remove doc table row
         await self.table_store.get_graph_embeddings(
             request.user,
             request.id,
-            publish
+            publish_ge,
         )
 
         print("Fetch complete", flush=True)
 
-        return KnowledgeResponse(
+        await respond(
-            error = None,
+            KnowledgeResponse(
-            ids = None,
+                error = None,
-            eos = True,
+                ids = None,
-            triples = None,
+                eos = True,
-            graph_embeddings = None,
+                triples = None,
+                graph_embeddings = None,
+            )
         )
 
-    async def list_kg_cores(self, request, flow):
+    async def list_kg_cores(self, request, respond):
 
         ids = await self.table_store.list_kg_cores(request.user)
 
-        return KnowledgeResponse(
+        await respond(
-            error = None,
+            KnowledgeResponse(
-            ids = ids,
+                error = None,
-            eos = False,
+                ids = ids,
-            triples = None,
+                eos = False,
-            graph_embeddings = None
+                triples = None,
+                graph_embeddings = None
+            )
         )
 

trustgraph-flow/trustgraph/cores/service.py

@@ -116,72 +116,7 @@ class Processor(AsyncProcessor):
 
         print(self.flows)
 
-    def __del__(self):
+    async def process_request(self, v, id):
-
-        pass
-
-    async def load_document(self, document, processing, content):
-
-        print("Ready for processing...")
-
-        print(document, processing, len(content))
-
-        if processing.flow not in self.flows:
-            raise RuntimeError("Invalid flow ID")
-
-        flow = self.flows[processing.flow]
-
-        if document.kind == "text/plain":
-            kind = "text-load"
-        elif document.kind == "application/pdf":
-            kind = "document-load"
-        else:
-            raise RuntimeError("Document with a MIME type I don't know")
-
-        q = flow["interfaces"][kind]
-
-        if kind == "text-load":
-            doc = TextDocument(
-                metadata = Metadata(
-                    id = document.id,
-                    metadata = document.metadata,
-                    user = processing.user,
-                    collection = processing.collection
-                ),
-                text = content,
-            )
-            schema = TextDocument
-        else:
-            doc = Document(
-                metadata = Metadata(
-                    id = document.id,
-                    metadata = document.metadata,
-                    user = processing.user,
-                    collection = processing.collection
-                ),
-                data = base64.b64encode(content).decode("utf-8")
-
-            )
-            schema = Document
-
-        print(f"Submit on queue {q}...")
-
-        pub = Publisher(
-            self.pulsar_client, q, schema=schema
-        )
-
-        await pub.start()
-
-        # FIXME: Time wait kludge?
-        await asyncio.sleep(1)
-
-        await pub.send(None, doc)
-
-        await pub.stop()
-
-        print("Document submitted")
-
-    async def process_request(self, v, flow):
 
         if v.operation is None:
             raise RequestError("Null operation")
@@ -197,7 +132,11 @@ class Processor(AsyncProcessor):
         if v.operation not in impls:
             raise RequestError(f"Invalid operation: {v.operation}")
 
-        return await impls[v.operation](v, flow)
+        async def respond(x):
+            await self.knowledge_response_producer.send(
+                x, { "id": id }
+            )
+        return await impls[v.operation](v, respond)
 
     async def on_knowledge_request(self, msg, consumer, flow):
 
@@ -211,11 +150,11 @@ class Processor(AsyncProcessor):
 
         try:
 
-            resp = await self.process_request(v, flow)
+            await self.process_request(v, id)
 
-            await self.knowledge_response_producer.send(
+#            await self.knowledge_response_producer.send(
-                resp, properties={"id": id}
+#                resp, properties={"id": id}
-            )
+#            )
 
             return
 

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

@@ -34,22 +34,34 @@ class KnowledgeRequestor(ServiceRequestor):
 
     def from_response(self, message):
 
-        print(message)
+        print("Processing message")
-
-        response = {
-            "eos": message.eos
-        }
 
+        # Response to list, 
         if message.ids is not None:
-            response["ids"] = message.ids
+            print("-> IDS")
+            return {
+                "ids": message.ids
+            }, True
 
         if message.triples:
-            response["triples"] = serialize_triples(message.triples)
+            print("-> triples")
+            return {
+                "triples": serialize_triples(message.triples)
+            }, False
 
         if message.graph_embeddings:
-            response["graph-embeddings"] = serialize_graph_embeddings(
+            print("-> ge")
-                message.graph_embeddings
+            return {
-            )
+                "graph-embeddings": serialize_graph_embeddings(
-        
+                    message.graph_embeddings
-        return response, True
+                )
+            }, False
+
+        if message.eos is True:
+            print("-> eos")
+            return {
+                "eos": True
+            }, True
+
+        raise RuntimeError("Unexpected case")
 

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

@@ -82,7 +82,7 @@ class ServiceRequestor:
 
                 resp, fin = self.from_response(resp)
 
-                print(resp, fin)
+                print(resp, fin, flush=True)
 
                 if responder:
                     await responder(resp, fin)

trustgraph-flow/trustgraph/tables/knowledge.py

@@ -351,17 +351,10 @@ class KnowledgeTableStore:
 
             try:
 
-                print("Executing!")
-                print(self.get_triples_stmt)
-                print(user, document_id)
-
                 resp = self.cassandra.execute(
                     self.get_triples_stmt,
                     (user, document_id)
                 )
-                print("HERE")
-
-                print("Executed")
 
                 break
 
@@ -370,12 +363,8 @@ class KnowledgeTableStore:
                 print(f"{e}, retry...", flush=True)
                 await asyncio.sleep(1)
 
-        print("Unpack...")
         for row in resp:
 
-            print("ROW")
-            print(row)
-
             if row[2]:
                 metadata = [
                     Triple(
@@ -402,6 +391,7 @@ class KnowledgeTableStore:
                     metadata = Metadata(
                         id = document_id,
                         user = user,
+                        collection = "default",  # FIXME: What to put here?
                         metadata = metadata,
                     ),
                     triples = triples
@@ -432,7 +422,6 @@ class KnowledgeTableStore:
 
         for row in resp:
 
-            print("MD...")
             if row[2]:
                 metadata = [
                     Triple(
@@ -445,7 +434,6 @@ class KnowledgeTableStore:
             else:
                 metadata = []
 
-            print("EE...")
             entities = [
                 EntityEmbeddings(
                     entity = Value(value = ent[0][0], is_uri = ent[0][1]),
@@ -459,6 +447,7 @@ class KnowledgeTableStore:
                     metadata = Metadata(
                         id = document_id,
                         user = user,
+                        collection = "default",   # FIXME: What to put here?
                         metadata = metadata,
                     ),
                     entities = entities