Python操作Neo4j图数据库的两种方式
前言
正在学习和使用知识图谱，先弄明白工具，图数据库neo4j由Java编写，但也有Python driver。
两种方式：
1、执行CQL ( cypher ) 语句
2、通过操作Python变量，达到操作neo4j的目的
本文包含三部分：
1、（用neo4j模块）执行CQL ( cypher ) 语句
2、（用py2neo模块）通过操作python变量，达到操作neo4j的目的
3、（用py2neo模块）执行CQL ( cypher ) 语句
py2neo比较方便：
-
优点：符合python的习惯，写着感觉顺畅，其实可以完全不会CQL也能写
-
缺点：代码长度比纯CQL要长，熟悉CQL的人可能会感觉拖沓
1. 用neo4j模块执行CQL ( cypher ) 语句
CQL(cypher)语法快查：官方Cyper语法
官方示例1:
from neo4j import GraphDatabase
driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
def add_friend(tx, name, friend_name):
tx.run("MERGE (a:Person {name: $name}) "
"MERGE (a)-[:KNOWS]->(friend:Person {name: $friend_name})",
name=name, friend_name=friend_name)
def print_friends(tx, name):
for record in tx.run("MATCH (a:Person)-[:KNOWS]->(friend) WHERE a.name = $name "
"RETURN friend.name ORDER BY friend.name", name=name):
print(record["friend.name"])
with driver.session() as session:
session.write_transaction(add_friend, "Arthur", "Guinevere")
session.write_transaction(add_friend, "Arthur", "Lancelot")
session.write_transaction(add_friend, "Arthur", "Merlin")
session.read_transaction(print_friends, "Arthur")
官方示例2：
上述程序的核心部分，抽象一下就是：
neo4j.GraphDatabase.driver(xxxx).session().write_transaction(函数(含tx.run(CQL语句)))
或
neo4j.GraphDatabase.driver(xxxx).session().begin_transaction.run(CQL语句)
附一个挺好的程序，可以直接用的：
# !/usr/bin/python
# -*- coding: utf-8 -*-
"""
create_author : 蛙鳜鸡鹳狸猿
create_time : 2017-04-18
program : *_* Read and Write Neo4j *_*
"""
from neo4j.v1 import GraphDatabase
class Neo4jHandler:
"""
Handler of graph database Neo4j reading and writing.
"""
def __init__(self, driver):
"""
Get Neo4j server driver.
:param driver: driver object
A driver object holds the detail of a Neo4j database including server URIs, credentials and other configuration, see
" http://neo4j.com/docs/api/python-driver/current/driver.html ".
"""
self.driver = driver
def __repr__(self):
printer = 'o(>﹏<)o ......Neo4j old driver "{0}" carry me fly...... o(^o^)o'.format(self.driver)
return printer
def listreader(self, cypher, keys):
"""
Read data from Neo4j in specified cypher.
Read and parse data straightly from cypher field result.
:param cypher: string
Valid query cypher statement.
:param keys: list
Cypher query columns to return.
:return: list
Each returned record constructs a list and stored in a big list, [[...], [...], ...].
"""
with self.driver.session() as session:
with session.begin_transaction() as tx:
data = []
result = tx.run(cypher)
for record in result:
rows = []
for key in keys:
rows.append(record[key])
data.append(rows)
return data
def dictreader(self, cypher):
"""
Read data from Neo4j in specified cypher.
The function depends on constructing dict method of dict(key = value) and any error may occur if the "key" is invalid to Python.
you can choose function dictreaderopted() below to read data by hand(via the args "keys").
:param cypher: string
Valid query cypher statement.
:return: list
Each returned record constructs a dict in "key : value" pairs and stored in a big list, [{...}, {...}, ...].
"""
with self.driver.session() as session:
with session.begin_transaction() as tx:
data = []
for record in tx.run(cypher).records():
item = {}
for args in str(record).split('>')[0].split()[1:]:
exec "item.update(dict({0}))".format(args)
data.append(item)
return data
def dictreaderopted(self, cypher, keys=None):
"""
Optimized function of dictreader().
Read and parse data straightly from cypher field result.
:param cypher: string
Valid query cypher statement.
:param keys: list, default : none(call dictreader())
Cypher query columns to return.
:return: list.
Each returned record constructs an dict in "key : value" pairs and stored in a list, [{...}, {...}, ...].
"""
if not keys:
return self.dictreader(cypher)
else:
with self.driver.session() as session:
with session.begin_transaction() as tx:
data = []
result = tx.run(cypher)
for record in result:
item = {}
for key in keys:
item.update({key : record[key]})
data.append(item)
return data
def cypherexecuter(self, cypher):
"""
Execute manipulation into Neo4j in specified cypher.
:param cypher: string
Valid handle cypher statement.
:return: none.
"""
with self.driver.session() as session:
with session.begin_transaction() as tx:
tx.run(cypher)
session.close()
# self test
if __name__ == "__main__":
uri = "bolt://localhost:7687"
driver = GraphDatabase.driver(uri, auth=("neo4j", "520"))
MyNH = Neo4jHandler(driver)
print(MyNH)
cypher_exec = """
CREATE (Neo:Crew {name:'Neo'}),
(Morpheus:Crew {name: 'Morpheus'}),
(Trinity:Crew {name: 'Trinity'}),
(Cypher:Crew:Matrix {name: 'Cypher'}),
(Smith:Matrix {name: 'Agent Smith'}),
(Architect:Matrix {name:'The Architect'}),
(Neo)-[:KNOWS]->(Morpheus),
(Neo)-[:LOVES]->(Trinity),
(Morpheus)-[:KNOWS]->(Trinity),
(Morpheus)-[:KNOWS]->(Cypher),
(Cypher)-[:KNOWS]->(Smith),
(Smith)-[:CODED_BY]->(Architect)
""" # "example cypher statement from http://console.neo4j.org/"
cypher_read = """
MATCH (a) -[:KNOWS|LOVES]-> (b:Crew {name: 'Trinity'})
RETURN a.name AS l, b.name AS r
"""
MyNH.cypherexecuter(cypher_exec)
print(MyNH.listreader(cypher_read, ['l', 'r']))
print(MyNH.dictreader(cypher_read))
print(MyNH.dictreaderopted(cypher_read, ['l']))
原文来源：Link
2. 用py2neo模块通过操作python变量，达到操作neo4j的目的
示例：
from py2neo import Graph, Node, Relationship
g = Graph()
tx = g.begin()
a = Node("Person", name="Alice")
tx.create(a)
b = Node("Person", name="Bob")
ab = Relationship(a, "KNOWS", b)
tx.create(ab)
tx.commit()
一个很好的介绍，基本操作都在里面：
写的时候可以先把基本操作做成函数，或者封装成类，方便使用。
3. 用py2neo模块执行CQL ( cypher ) 语句
直接看例子：Link
import json
from bottle import get, run, request, response, static_file
from py2neo import Graph
#password = {Your neo4j password}
graph = Graph(password = "xxxxxx")
@get("/")
def get_index():
return static_file("index.html", root="static")
@get("/graph")
def get_graph():
results = graph.run(
"MATCH (m:Movie)<-[:ACTED_IN]-(a:Person) "
"RETURN m.title as movie, collect(a.name) as cast "
"LIMIT {limit}", {"limit": 100})
nodes = []
rels = []
i = 0
for movie, cast in results:
nodes.append({"title": movie, "label": "movie"})
target = i
i += 1
for name in cast:
actor = {"title": name, "label": "actor"}
try:
source = nodes.index(actor)
except ValueError:
nodes.append(actor)
source = i
i += 1
rels.append({"source": source, "target": target})
return {"nodes": nodes, "links": rels}
@get("/search")
def get_search():
try:
q = request.query["q"]
except KeyError:
return []
else:
results = graph.run(
"MATCH (movie:Movie) "
"WHERE movie.title =~ {title} "
"RETURN movie", {"title": "(?i).*" + q + ".*"})
response.content_type = "application/json"
return json.dumps([{"movie": dict(row["movie"])} for row in results])
@get("/movie/<title>")
def get_movie(title):
results = graph.run(
"MATCH (movie:Movie {title:{title}}) "
"OPTIONAL MATCH (movie)<-[r]-(person:Person) "
"RETURN movie.title as title,"
"collect([person.name, head(split(lower(type(r)),'_')), r.roles]) as cast "
"LIMIT 1", {"title": title})
row = results.next()
return {"title": row["title"],
"cast": [dict(zip(("name", "job", "role"), member)) for member in row["cast"]]}
if __name__ == "__main__":
run(port=8080)
其中核心部分抽象就是：
py2neo.Graph(xxxx).run(CQL语句) # 返回一个二维结果
参考：Link
加油!
感谢!
努力!