- Added triple query using triples-query-cassandra

- Added triples_query_client.py and a test

scripts/triples-query-cassandra

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.query.triples.cassandra import run
+
+run()
+

tests/test-graph-embeddings

@@ -20,4 +20,3 @@ print("Response...")
 for val in resp:
     print(val.value)
 
-

tests/test-triples

@@ -0,0 +1,70 @@
+#!/usr/bin/env python3
+
+import pulsar
+from trustgraph.triples_query_client import TriplesQueryClient
+
+tq = TriplesQueryClient(pulsar_host="pulsar://localhost:6650")
+
+e = "http://trustgraph.ai/e/shuttle"
+
+limit=3
+
+def dump(resp):
+    print("Response...")
+    for t in resp:
+        print(t.s.value, t.p.value, t.o.value)
+
+print("-- * ---------------------------")
+
+resp = tq.request(None, None, None, limit)
+dump(resp)
+
+print("-- s ---------------------------")
+
+resp = tq.request("http://trustgraph.ai/e/shuttle", None, None, limit)
+dump(resp)
+
+print("-- p ---------------------------")
+
+resp = tq.request(None, "http://trustgraph.ai/e/landed", None, limit)
+dump(resp)
+
+print("-- p ---------------------------")
+
+resp = tq.request(None, None, "President", limit)
+dump(resp)
+
+print("-- sp ---------------------------")
+
+resp = tq.request(
+    "http://trustgraph.ai/e/shuttle", "http://trustgraph.ai/e/landed", None,
+    limit
+)
+dump(resp)
+
+print("-- so ---------------------------")
+
+resp = tq.request(
+    "http://trustgraph.ai/e/shuttle", None, "the tower",
+    limit
+)
+dump(resp)
+
+print("-- po ---------------------------")
+
+resp = tq.request(
+    "http://trustgraph.ai/e/shuttle", "http://trustgraph.ai/e/landed",
+    None,
+    limit
+)
+dump(resp)
+
+print("-- spo ---------------------------")
+
+resp = tq.request(
+    "http://trustgraph.ai/e/shuttle", "http://trustgraph.ai/e/landed",
+    "on the concrete runway at Kennedy Space Center",
+    limit
+)
+dump(resp)
+

trustgraph/direct/cassandra.py

@@ -67,7 +67,7 @@ class TrustGraph:
 
     def get_s(self, s, limit=10):
         return self.session.execute(
-            f"select p, o from triples where s = %s",
+            f"select p, o from triples where s = %s limit {limit}",
             (s,)
         )
 
@@ -97,7 +97,7 @@ class TrustGraph:
 
     def get_os(self, o, s, limit=10):
         return self.session.execute(
-            f"select s from triples where o = %s and s = %s limit {limit}",
+            f"select p from triples where o = %s and s = %s limit {limit}",
             (o, s)
         )
 

trustgraph/query/graph_embeddings/milvus/service.py

@@ -1,7 +1,7 @@
 
 """
 Graph embeddings query service.  Input is vector, output is list of
-e
+entities
 """
 
 from .... direct.milvus import TripleVectors
@@ -39,6 +39,12 @@ class Processor(ConsumerProducer):
 
         self.vecstore = TripleVectors(store_uri)
 
+    def create_value(self, ent):
+        if ent.startswith("http://") or ent.startswith("https://"):
+            return Value(value=ent, is_uri=True)
+        else:
+            return Value(value=ent, is_uri=False)
+        
     def handle(self, msg):
 
         v = msg.value()
@@ -64,10 +70,7 @@ class Processor(ConsumerProducer):
         ents2 = []
 
         for ent in entities:
-            if ent.startswith("http://") or ent.startswith("https://"):
-                ents2.append(Value(value=ent, is_uri=True))
-            else:
-                ents2.append(Value(value=ent, is_uri=False))
+            ents2.append(self.create_value(ent))
 
         entities = ents2
 

trustgraph/query/triples/cassandra/__init__.py

@@ -0,0 +1,3 @@
+
+from . service import *
+

trustgraph/query/triples/cassandra/__main__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+
+from . hf import run
+
+if __name__ == '__main__':
+    run()
+

trustgraph/query/triples/cassandra/service.py

@@ -0,0 +1,153 @@
+
+"""
+Triples query service.  Input is a (s, p, o) triple, some values may be
+null.  Output is a list of triples.
+"""
+
+from .... direct.cassandra import TrustGraph
+from .... schema import TriplesQueryRequest, TriplesQueryResponse
+from .... schema import Value, Triple
+from .... schema import triples_request_queue
+from .... schema import triples_response_queue
+from .... base import ConsumerProducer
+
+module = ".".join(__name__.split(".")[1:-1])
+
+default_input_queue = triples_request_queue
+default_output_queue = triples_response_queue
+default_subscriber = module
+default_graph_host='localhost'
+
+class Processor(ConsumerProducer):
+
+    def __init__(self, **params):
+
+        input_queue = params.get("input_queue", default_input_queue)
+        output_queue = params.get("output_queue", default_output_queue)
+        subscriber = params.get("subscriber", default_subscriber)
+        graph_host = params.get("graph_host", default_graph_host)
+
+        super(Processor, self).__init__(
+            **params | {
+                "input_queue": input_queue,
+                "output_queue": output_queue,
+                "subscriber": subscriber,
+                "input_schema": TriplesQueryRequest,
+                "output_schema": TriplesQueryResponse,
+                "graph_host": graph_host,
+            }
+        )
+
+        self.tg = TrustGraph([graph_host])
+
+    def create_value(self, ent):
+        if ent.startswith("http://") or ent.startswith("https://"):
+            return Value(value=ent, is_uri=True)
+        else:
+            return Value(value=ent, is_uri=False)
+
+    def handle(self, msg):
+
+        v = msg.value()
+
+        # Sender-produced ID
+        id = msg.properties()["id"]
+
+        print(f"Handling input {id}...", flush=True)
+
+        triples = []
+
+        if v.s is not None:
+            if v.p is not None:
+                if v.o is not None:
+                    resp = self.tg.get_spo(
+                        v.s.value, v.p.value, v.o.value,
+                        limit=v.limit
+                    )
+                    triples.append((v.s.value, v.p.value, v.o.value))
+                else:
+                    resp = self.tg.get_sp(
+                        v.s.value, v.p.value,
+                        limit=v.limit
+                    )
+                    for t in resp:
+                        triples.append((v.s.value, v.p.value, t.o))
+            else:
+                if v.o is not None:
+                    resp = self.tg.get_os(
+                        v.o.value, v.s.value, 
+                        limit=v.limit
+                    )
+                    for t in resp:
+                        triples.append((v.s.value, t.p, v.o.value))
+                else:
+                    resp = self.tg.get_s(
+                        v.s.value,
+                        limit=v.limit
+                    )
+                    for t in resp:
+                        triples.append((v.s.value, t.p, t.o))
+        else:
+            if v.p is not None:
+                if v.o is not None:
+                    resp = self.tg.get_po(
+                        v.p.value, v.o.value,
+                        limit=v.limit
+                    )
+                    for t in resp:
+                        triples.append((t.s, v.p.value, v.o.value))
+                else:
+                    resp = self.tg.get_p(
+                        v.p.value,
+                        limit=v.limit
+                    )
+                    for t in resp:
+                        triples.append((t.s, v.p.value, t.o))
+            else:
+                if v.o is not None:
+                    resp = self.tg.get_o(
+                        v.o.value,
+                        limit=v.limit
+                    )
+                    for t in resp:
+                        triples.append((t.s, t.p, v.o.value))
+                else:
+                    resp = self.tg.get_all(
+                        limit=v.limit
+                    )
+                    for t in resp:
+                        triples.append((t.s, t.p, t.o))
+
+        triples = [
+            Triple(
+                s=self.create_value(t[0]),
+                p=self.create_value(t[1]), 
+                o=self.create_value(t[2])
+            )
+            for t in triples
+        ]
+
+        print("Send response...", flush=True)
+        r = TriplesQueryResponse(triples=triples)
+        self.producer.send(r, properties={"id": id})
+
+        print("Done.", flush=True)
+
+    @staticmethod
+    def add_args(parser):
+
+        ConsumerProducer.add_args(
+            parser, default_input_queue, default_subscriber,
+            default_output_queue,
+        )
+
+        parser.add_argument(
+            '-g', '--graph-host',
+            default="localhost",
+            help=f'Graph host (default: localhost)'
+        )
+
+def run():
+
+    Processor.start(module, __doc__)
+

trustgraph/storage/triples/cassandra/write.py

@@ -34,6 +34,7 @@ class Processor(Consumer):
                 "input_queue": input_queue,
                 "subscriber": subscriber,
                 "input_schema": Triple,
+                "graph_host": graph_host,
             }
         )
 

trustgraph/triples_query_client.py

@@ -0,0 +1,100 @@
+#!/usr/bin/env python3
+
+import pulsar
+import _pulsar
+from pulsar.schema import JsonSchema
+import hashlib
+import uuid
+
+from . schema import TriplesQueryRequest, TriplesQueryResponse, Value
+from . schema import triples_request_queue
+from . schema import triples_response_queue
+
+# Ugly
+ERROR=_pulsar.LoggerLevel.Error
+WARN=_pulsar.LoggerLevel.Warn
+INFO=_pulsar.LoggerLevel.Info
+DEBUG=_pulsar.LoggerLevel.Debug
+
+class TriplesQueryClient:
+
+    def __init__(
+            self, log_level=ERROR,
+            subscriber=None,
+            input_queue=None,
+            output_queue=None,
+            pulsar_host="pulsar://pulsar:6650",
+    ):
+
+        if input_queue == None:
+            input_queue = triples_request_queue
+
+        if output_queue == None:
+            output_queue = triples_response_queue
+
+        if subscriber == None:
+            subscriber = str(uuid.uuid4())
+
+        self.client = pulsar.Client(
+            pulsar_host,
+            logger=pulsar.ConsoleLogger(log_level),
+        )
+
+        self.producer = self.client.create_producer(
+            topic=input_queue,
+            schema=JsonSchema(TriplesQueryRequest),
+            chunking_enabled=True,
+        )
+
+        self.consumer = self.client.subscribe(
+            output_queue, subscriber,
+            schema=JsonSchema(TriplesQueryResponse),
+        )
+
+    def create_value(self, ent):
+
+        if ent == None: return None
+
+        if ent.startswith("http://") or ent.startswith("https://"):
+            return Value(value=ent, is_uri=True)
+
+        return Value(value=ent, is_uri=False)
+
+    def request(self, s, p, o, limit=10, timeout=500):
+
+        id = str(uuid.uuid4())
+
+        r = TriplesQueryRequest(
+            s=self.create_value(s),
+            p=self.create_value(p),
+            o=self.create_value(o),
+            limit=limit,
+        )
+
+        self.producer.send(r, properties={ "id": id })
+
+        while True:
+
+            msg = self.consumer.receive(timeout_millis=timeout * 1000)
+
+            mid = msg.properties()["id"]
+
+            if mid == id:
+                resp = msg.value().triples
+                self.consumer.acknowledge(msg)
+                return resp
+
+            # Ignore messages with wrong ID
+            self.consumer.acknowledge(msg)
+
+    def __del__(self):
+
+        if hasattr(self, "consumer"):
+            self.consumer.close()
+            
+        if hasattr(self, "producer"):
+            self.producer.flush()
+            self.producer.close()
+            
+        self.client.close()
+