apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-07 04:12:10 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Some querying functions work on Neo

Browse source
This commit is contained in:
Cyber MacGeddon 2024-08-13 23:57:48 +01:00
parent 053fc5511b
commit 737bedc62a
3 changed files with 143 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

4
tests/test-triples
View file

@ -53,8 +53,8 @@ dump(resp)
print("-- po ---------------------------") print("-- po ---------------------------")
resp = tq.request( resp = tq.request(
"http://trustgraph.ai/e/shuttle", "http://trustgraph.ai/e/landed", None, "http://trustgraph.ai/e/landed",
None, "on the concrete runway at Kennedy Space Center",
limit limit
) )
dump(resp) dump(resp)

187
trustgraph/query/triples/neo4j/service.py
View file

@ -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. null. Output is a list of triples.
""" """
from .... direct.cassandra import TrustGraph from neo4j import GraphDatabase
from .... schema import TriplesQueryRequest, TriplesQueryResponse from .... schema import TriplesQueryRequest, TriplesQueryResponse
from .... schema import Value, Triple from .... schema import Value, Triple
from .... schema import triples_request_queue from .... schema import triples_request_queue
@ -16,7 +17,10 @@ module = ".".join(__name__.split(".")[1:-1])
default_input_queue = triples_request_queue default_input_queue = triples_request_queue
default_output_queue = triples_response_queue default_output_queue = triples_response_queue
default_subscriber = module default_subscriber = module
default_graph_host='localhost'
default_graph_host = 'bolt://localhost:7687'
default_username = 'neo4j'
default_password = 'password'
class Processor(ConsumerProducer): class Processor(ConsumerProducer):
@ -26,6 +30,8 @@ class Processor(ConsumerProducer):
output_queue = params.get("output_queue", default_output_queue) output_queue = params.get("output_queue", default_output_queue)
subscriber = params.get("subscriber", default_subscriber) subscriber = params.get("subscriber", default_subscriber)
graph_host = params.get("graph_host", default_graph_host) 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__( super(Processor, self).__init__(
**params | { **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): def create_value(self, ent):
if ent.startswith("http://") or ent.startswith("https://"): if ent.startswith("http://") or ent.startswith("https://"):
return Value(value=ent, is_uri=True) return Value(value=ent, is_uri=True)
else: else:
@ -60,63 +69,135 @@ class Processor(ConsumerProducer):
if v.s is not None: if v.s is not None:
if v.p is not None: if v.p is not None:
if v.o is not None: if v.o is not None:
resp = self.tg.get_spo(
v.s.value, v.p.value, v.o.value, # SPO
limit=v.limit
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: else:
resp = self.tg.get_sp(
v.s.value, v.p.value, # SP
limit=v.limit
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: else:
if v.o is not None: if v.o is not None:
resp = self.tg.get_os(
v.o.value, v.s.value, # SO
limit=v.limit
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: else:
resp = self.tg.get_s(
v.s.value, # S
limit=v.limit
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: else:
if v.p is not None: if v.p is not None:
if v.o is not None: if v.o is not None:
resp = self.tg.get_po(
v.p.value, v.o.value, # PO
limit=v.limit pass
)
for t in resp:
triples.append((t.s, v.p.value, v.o.value))
else: else:
resp = self.tg.get_p(
v.p.value, # P
limit=v.limit pass
)
for t in resp:
triples.append((t.s, v.p.value, t.o))
else: else:
if v.o is not None: if v.o is not None:
resp = self.tg.get_o(
v.o.value, # O
limit=v.limit pass
)
for t in resp:
triples.append((t.s, t.p, v.o.value))
else: else:
resp = self.tg.get_all(
limit=v.limit # *
) pass
for t in resp:
triples.append((t.s, t.p, t.o))
triples = [ triples = [
Triple( Triple(
@ -143,8 +224,20 @@ class Processor(ConsumerProducer):
parser.add_argument( parser.add_argument(
'-g', '--graph-host', '-g', '--graph-host',
default="localhost", default=default_graph_host,
help=f'Graph host (default: localhost)' 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(): def run():

1
trustgraph/storage/triples/neo4j/write.py
View file

@ -111,6 +111,7 @@ class Processor(Consumer):
nodes_created=summary.counters.nodes_created, nodes_created=summary.counters.nodes_created,
time=summary.result_available_after time=summary.result_available_after
)) ))
def handle(self, msg): def handle(self, msg):
v = msg.value() v = msg.value()