From ee89a5d8e6807dd8d222efec3005618d9396f62d Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Mon, 12 Aug 2024 20:12:13 +0100 Subject: [PATCH] - Added triple query using triples-query-cassandra - Added triples_query_client.py and a test --- scripts/triples-query-cassandra | 6 + tests/test-graph-embeddings | 1 - tests/test-triples | 70 ++++++++ trustgraph/direct/cassandra.py | 4 +- .../query/graph_embeddings/milvus/service.py | 13 +- trustgraph/query/triples/__init__.py | 0 .../query/triples/cassandra/__init__.py | 3 + .../query/triples/cassandra/__main__.py | 7 + trustgraph/query/triples/cassandra/service.py | 153 ++++++++++++++++++ trustgraph/storage/triples/cassandra/write.py | 1 + trustgraph/triples_query_client.py | 100 ++++++++++++ 11 files changed, 350 insertions(+), 8 deletions(-) create mode 100755 scripts/triples-query-cassandra create mode 100755 tests/test-triples create mode 100644 trustgraph/query/triples/__init__.py create mode 100644 trustgraph/query/triples/cassandra/__init__.py create mode 100755 trustgraph/query/triples/cassandra/__main__.py create mode 100755 trustgraph/query/triples/cassandra/service.py create mode 100644 trustgraph/triples_query_client.py diff --git a/scripts/triples-query-cassandra b/scripts/triples-query-cassandra new file mode 100755 index 00000000..d6baf969 --- /dev/null +++ b/scripts/triples-query-cassandra @@ -0,0 +1,6 @@ +#!/usr/bin/env python3 + +from trustgraph.query.triples.cassandra import run + +run() + diff --git a/tests/test-graph-embeddings b/tests/test-graph-embeddings index 742e7f81..188ee6e5 100755 --- a/tests/test-graph-embeddings +++ b/tests/test-graph-embeddings @@ -20,4 +20,3 @@ print("Response...") for val in resp: print(val.value) - diff --git a/tests/test-triples b/tests/test-triples new file mode 100755 index 00000000..9762d672 --- /dev/null +++ b/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) + diff --git a/trustgraph/direct/cassandra.py b/trustgraph/direct/cassandra.py index cb313abc..1754e090 100644 --- a/trustgraph/direct/cassandra.py +++ b/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) ) diff --git a/trustgraph/query/graph_embeddings/milvus/service.py b/trustgraph/query/graph_embeddings/milvus/service.py index c95f2ef5..74e0bee8 100755 --- a/trustgraph/query/graph_embeddings/milvus/service.py +++ b/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 diff --git a/trustgraph/query/triples/__init__.py b/trustgraph/query/triples/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/trustgraph/query/triples/cassandra/__init__.py b/trustgraph/query/triples/cassandra/__init__.py new file mode 100644 index 00000000..ba844705 --- /dev/null +++ b/trustgraph/query/triples/cassandra/__init__.py @@ -0,0 +1,3 @@ + +from . service import * + diff --git a/trustgraph/query/triples/cassandra/__main__.py b/trustgraph/query/triples/cassandra/__main__.py new file mode 100755 index 00000000..89684e3e --- /dev/null +++ b/trustgraph/query/triples/cassandra/__main__.py @@ -0,0 +1,7 @@ +#!/usr/bin/env python3 + +from . hf import run + +if __name__ == '__main__': + run() + diff --git a/trustgraph/query/triples/cassandra/service.py b/trustgraph/query/triples/cassandra/service.py new file mode 100755 index 00000000..c40a50fc --- /dev/null +++ b/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__) + diff --git a/trustgraph/storage/triples/cassandra/write.py b/trustgraph/storage/triples/cassandra/write.py index 5ba657aa..84c002ff 100755 --- a/trustgraph/storage/triples/cassandra/write.py +++ b/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, } ) diff --git a/trustgraph/triples_query_client.py b/trustgraph/triples_query_client.py new file mode 100644 index 00000000..0a97d66d --- /dev/null +++ b/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() +