-
Notifications
You must be signed in to change notification settings - Fork 0
/
learn.py
62 lines (53 loc) · 2.6 KB
/
learn.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
from neo4j import GraphDatabase
import logging
from neo4j.exceptions import ServiceUnavailable
class App:
def __init__(self, uri, user, password):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def close(self):
# Don't forget to close the driver connection when you are finished with it
self.driver.close()
def create_friendship(self, person1_name, person2_name):
with self.driver.session() as session:
# Write transactions allow the driver to handle retries and transient errors
result = session.write_transaction(self._create_and_return_friendship, person1_name, person2_name)
for row in result:
print("Created friendship between: {p1}, {p2}".format(p1=row['p1'], p2=row['p2']))
@staticmethod
def _create_and_return_friendship(tx, person1_name, person2_name):
# To learn more about the Cypher syntax, see https://neo4j.com/docs/cypher-manual/current/
# The Reference Card is also a good resource for keywords https://neo4j.com/docs/cypher-refcard/current/
query = (
"CREATE (p1:Person { name: $person1_name }) "
"CREATE (p2:Person { name: $person2_name }) "
"CREATE (p1)-[:KNOWS]->(p2) "
"RETURN p1, p2"
)
result = tx.run(query, person1_name=person1_name, person2_name=person2_name)
try:
return [{"p1": row["p1"]["name"], "p2": row["p2"]["name"]} for row in result]
# Capture any errors along with the query and data for traceability
except ServiceUnavailable as exception:
logging.error("{query} raised an error: \n {exception}".format(
query=query, exception=exception))
raise
def find_person(self, person_name):
with self.driver.session() as session:
result = session.read_transaction(self._find_and_return_person, person_name)
for row in result:
print("Found person: {row}".format(row=row))
@staticmethod
def _find_and_return_person(tx, person_name):
query = (
"MATCH (p:Person) "
"WHERE p.name = $person_name "
"RETURN p.name AS name"
)
result = tx.run(query, person_name=person_name)
return [row["name"] for row in result]
if __name__ == "__main__":
# Aura queries use an encrypted connection using the "neo4j+s" URI scheme
app = App("bolt://localhost:7687", "neo4j", "sysmon")
app.create_friendship("Alice", "David")
app.find_person("Alice")
app.close()