From 737bedc62a9b996b10abdbdffd153f2113ee8ded Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Tue, 13 Aug 2024 23:57:48 +0100 Subject: [PATCH] Some querying functions work on Neo --- tests/test-triples | 4 +- trustgraph/query/triples/neo4j/service.py | 187 ++++++++++++++++------ trustgraph/storage/triples/neo4j/write.py | 1 + 3 files changed, 143 insertions(+), 49 deletions(-) diff --git a/tests/test-triples b/tests/test-triples index 9762d672..bdd781f9 100755 --- a/tests/test-triples +++ b/tests/test-triples @@ -53,8 +53,8 @@ dump(resp) print("-- po ---------------------------") resp = tq.request( - "http://trustgraph.ai/e/shuttle", "http://trustgraph.ai/e/landed", - None, + None, "http://trustgraph.ai/e/landed", + "on the concrete runway at Kennedy Space Center", limit ) dump(resp) diff --git a/trustgraph/query/triples/neo4j/service.py b/trustgraph/query/triples/neo4j/service.py index c40a50fc..19d060fc 100755 --- a/trustgraph/query/triples/neo4j/service.py +++ b/trustgraph/query/triples/neo4j/service.py @@ -4,7 +4,8 @@ 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 neo4j import GraphDatabase + from .... schema import TriplesQueryRequest, TriplesQueryResponse from .... schema import Value, Triple from .... schema import triples_request_queue @@ -16,7 +17,10 @@ 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' + +default_graph_host = 'bolt://localhost:7687' +default_username = 'neo4j' +default_password = 'password' class Processor(ConsumerProducer): @@ -26,6 +30,8 @@ class Processor(ConsumerProducer): output_queue = params.get("output_queue", default_output_queue) subscriber = params.get("subscriber", default_subscriber) graph_host = params.get("graph_host", default_graph_host) + username = params.get("username", default_username) + password = params.get("passowrd", default_password) super(Processor, self).__init__( **params | { @@ -38,9 +44,12 @@ class Processor(ConsumerProducer): } ) - self.tg = TrustGraph([graph_host]) + self.db = "neo4j" + + self.io = GraphDatabase.driver(graph_host, auth=(username, password)) def create_value(self, ent): + if ent.startswith("http://") or ent.startswith("https://"): return Value(value=ent, is_uri=True) else: @@ -60,63 +69,135 @@ class Processor(ConsumerProducer): 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 + + # SPO + + records, summary, keys = self.io.execute_query( + "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Literal {value: $value}) " + "RETURN $src as src", + src=v.s.value, rel=v.p.value, value=v.o.value, + database_=self.db, ) - triples.append((v.s.value, v.p.value, v.o.value)) + + for rec in records: + triples.append((v.s.value, v.p.value, v.o.value)) + + records, summary, keys = self.io.execute_query( + "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Node {uri: $uri}) " + "RETURN $src as src", + src=v.s.value, rel=v.p.value, uri=v.o.value, + database_=self.db, + ) + + for rec in records: + 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 + + # SP + + records, summary, keys = self.io.execute_query( + "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Literal) " + "RETURN dest.value as dest", + src=v.s.value, rel=v.p.value, + database_=self.db, ) - for t in resp: - triples.append((v.s.value, v.p.value, t.o)) + + for rec in records: + data = rec.data() + triples.append((v.s.value, v.p.value, data["dest"])) + + records, summary, keys = self.io.execute_query( + "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Node) " + "RETURN dest.uri as dest", + src=v.s.value, rel=v.p.value, + database_=self.db, + ) + + for rec in records: + data = rec.data() + triples.append((v.s.value, v.p.value, data["dest"])) + else: + if v.o is not None: - resp = self.tg.get_os( - v.o.value, v.s.value, - limit=v.limit + + # SO + + records, summary, keys = self.io.execute_query( + "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Literal {value: $value}) " + "RETURN rel.uri as rel", + src=v.s.value, value=v.o.value, + database_=self.db, ) - for t in resp: - triples.append((v.s.value, t.p, v.o.value)) + + for rec in records: + data = rec.data() + triples.append((v.s.value, data["rel"], v.o.value)) + + records, summary, keys = self.io.execute_query( + "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Node {uri: $uri}) " + "RETURN rel.uri as rel", + src=v.s.value, uri=v.o.value, + database_=self.db, + ) + + for rec in records: + data = rec.data() + triples.append((v.s.value, data["rel"], v.o.value)) + else: - resp = self.tg.get_s( - v.s.value, - limit=v.limit + + # S + + records, summary, keys = self.io.execute_query( + "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Literal) " + "RETURN rel.uri as rel, dest.value as dest", + src=v.s.value, + database_=self.db, ) - for t in resp: - triples.append((v.s.value, t.p, t.o)) + + for rec in records: + data = rec.data() + triples.append((v.s.value, data["rel"], data["dest"])) + + records, summary, keys = self.io.execute_query( + "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Node) " + "RETURN rel.uri as rel, dest.uri as dest", + src=v.s.value, + database_=self.db, + ) + + for rec in records: + data = rec.data() + triples.append((v.s.value, data["rel"], data["dest"])) + + 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)) + + # PO + pass + 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)) + + # P + pass + 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)) + + # O + pass + else: - resp = self.tg.get_all( - limit=v.limit - ) - for t in resp: - triples.append((t.s, t.p, t.o)) + + # * + pass triples = [ Triple( @@ -143,8 +224,20 @@ class Processor(ConsumerProducer): parser.add_argument( '-g', '--graph-host', - default="localhost", - help=f'Graph host (default: localhost)' + default=default_graph_host, + help=f'Graph host (default: {default_graph_host})' + ) + + parser.add_argument( + '--username', + default=default_username, + help=f'Neo4j username (default: {default_username})' + ) + + parser.add_argument( + '--password', + default=default_password, + help=f'Neo4j password (default: {default_password})' ) def run(): diff --git a/trustgraph/storage/triples/neo4j/write.py b/trustgraph/storage/triples/neo4j/write.py index b5f4d049..c2ea497d 100755 --- a/trustgraph/storage/triples/neo4j/write.py +++ b/trustgraph/storage/triples/neo4j/write.py @@ -111,6 +111,7 @@ class Processor(Consumer): nodes_created=summary.counters.nodes_created, time=summary.result_available_after )) + def handle(self, msg): v = msg.value()