Python操作Neo4j图数据库的两种方式

• 前言
• 1. 用neo4j模块执行CQL ( cypher ) 语句
• 2. 用py2neo模块通过操作python变量，达到操作neo4j的目的
• 3. 用py2neo模块执行CQL ( cypher ) 语句

正在学习和使用知识图谱，先弄明白工具，图数据库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的人可能会感觉拖沓

neo4j 官方文档

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：

link

from neo4j.v1 import GraphDatabase
 
uri = "bolt://localhost:7687"
driver = GraphDatabase.driver(uri, auth=("neo4j", "520"))
 
def cyphertx(cypher):
    with driver.session() as session:
        with session.begin_transaction() as tx:
            tx.run(cypher)
 
cypher = """
            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)
         """   # "cypher from http://console.neo4j.org/"
cyphertx(cypher)

上述程序的核心部分，抽象一下就是：

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

官网文档

示例：

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()

一个很好的介绍，基本操作都在里面：

py2neo的简单使用

写的时候可以先把基本操作做成函数，或者封装成类，方便使用。

直接看例子：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/