1. 什么是 Neo4j Python 驱动？ #

1.1 驱动简介 #

neo4j 是 Neo4j 图数据库的官方 Python 驱动，提供了与 Neo4j 数据库交互的完整接口。

为什么需要驱动？

想象一下，Neo4j 数据库是一个仓库，而 Python 程序是一个客户。客户不能直接进入仓库，需要通过一个"翻译员"（驱动）来：

• 理解客户的需求（Python 代码）
• 翻译成仓库能理解的语言（Cypher 查询）
• 将结果返回给客户（Python 对象）

驱动的主要特点：

• 官方支持：Neo4j 官方维护，稳定可靠
• 纯 Python 实现：易于安装和使用
• 现代化 API：简单直观的接口设计
• 高性能：通过 Bolt 协议与数据库通信，速度快

1.2 前置知识 #

在学习本教程之前，你需要掌握：

• Python 基础：变量、函数、类、异常处理
• Neo4j 基础：了解什么是节点、关系、属性
• Cypher 基础：基本的查询语句（MATCH、CREATE、RETURN 等）

1.3 安装驱动 #

# 使用 pip 安装 neo4j 驱动
pip install neo4j

验证安装：

# 导入 neo4j 库，如果没有错误说明安装成功
import neo4j
print(f"Neo4j 驱动版本：{neo4j.__version__}")

2. 连接数据库 #

2.1 基础连接 #

连接 Neo4j 数据库是使用驱动的第一步。我们需要提供数据库的地址和认证信息。

连接信息说明：

• URI：数据库的地址，格式为 neo4j://主机:端口 或 bolt://主机:端口
• 认证信息：用户名和密码，格式为元组 (用户名, 密码)
• 默认端口：Neo4j 的 Bolt 协议默认端口是 7687

完整示例：

# 导入 Neo4j 驱动
from neo4j import GraphDatabase

# 数据库连接信息
URI = "bolt://localhost:7687"
USERNAME = "neo4j"
PASSWORD = "12345678"

# 创建数据库驱动
driver = GraphDatabase.driver(URI, auth=(USERNAME, PASSWORD))

# 验证连接
try:
    driver.verify_connectivity()
    print(" 连接成功！")
except Exception as e:
    print(f" 连接失败：{e}")

# 关闭连接
driver.close()

重要提示： 如果连接失败，检查：

1. Neo4j 数据库是否已启动
2. 端口号是否正确（默认是 7687）
3. 用户名和密码是否正确
4. 防火墙是否阻止了连接

2.2 连接配置 #

在实际应用中，可能需要配置一些连接参数：

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "12345678")

# 连接配置
CONFIG = {
    "max_connection_pool_size": 50,  # 连接池最大连接数
    "connection_timeout": 15,         # 连接超时时间（秒）
    "encrypted": False                # 是否启用加密
}

# 创建带配置的驱动
driver = GraphDatabase.driver(URI, auth=AUTH, **CONFIG)

try:
    driver.verify_connectivity()
    print(" 连接成功！")
except Exception as e:
    print(f" 连接失败：{e}")

driver.close()

3. 执行查询 #

3.1 使用 execute_query 方法（推荐） #

execute_query() 是 Neo4j 驱动提供的最简单、最常用的查询方法。它会自动管理会话和事务。

方法特点：

• 自动管理：自动创建会话和事务，无需手动管理
• 简单易用：一行代码就能执行查询
• 参数化查询：支持参数化查询，安全且高效

基础查询示例：

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "12345678")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 执行简单查询
# 返回三个值：records（记录列表）、summary（查询摘要）、keys（结果键）
records, summary, keys = driver.execute_query(
    "RETURN 'Hello, Neo4j!' AS message"
)

# 处理查询结果
for record in records:
    print(record["message"])

driver.close()

3.2 参数化查询 #

使用参数化查询可以避免注入攻击，提高安全性和性能。

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "12345678")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 参数化查询：使用 $参数名 定义参数
records, summary, keys = driver.execute_query(
    "MATCH (a:Actor) WHERE a.born > $min_year RETURN a.name AS name, a.born AS born",
    min_year=1960  # 传递参数值
)

print("1960年后出生的演员：")
for record in records:
    print(f"  {record['name']}: {record['born']} 年")

driver.close()

3.3 创建数据 #

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "12345678")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 创建节点
records, summary, keys = driver.execute_query(
    """
    CREATE (a:Actor {name: $name, born: $born})
    RETURN a
    """,
    name="周星驰",
    born=1962
)

for record in records:
    print(f"创建节点：{record['a']}")

# 查看创建统计
print(f"创建的节点数：{summary.counters.nodes_created}")

driver.close()

创建关系示例：

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "12345678")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 创建关系
records, summary, keys = driver.execute_query(
    """
    MATCH (a:Actor {name: $actor_name})
    MATCH (m:Movie {title: $movie_title})
    CREATE (a)-[r:ACTED_IN {roles: $roles}]->(m)
    RETURN r
    """,
    actor_name="周星驰",
    movie_title="功夫",
    roles=["阿星"]
)

print(f"创建的关系数：{summary.counters.relationships_created}")

driver.close()

3.4 更新数据 #

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "12345678")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 更新节点属性
records, summary, keys = driver.execute_query(
    """
    MATCH (a:Actor {name: $name})
    SET a.born = $new_born
    RETURN a
    """,
    name="周星驰",
    new_born=1962
)

print(f"更新的属性数：{summary.counters.properties_set}")

driver.close()

3.5 删除数据 #

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "12345678")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 删除节点（DETACH DELETE 会删除节点及其所有关系）
records, summary, keys = driver.execute_query(
    """
    MATCH (a:Actor {name: $name})
    DETACH DELETE a
    """,
    name="测试演员"
)

print(f"删除的节点数：{summary.counters.nodes_deleted}")

driver.close()

3.6 使用 MERGE 避免重复 #

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "12345678")
driver = GraphDatabase.driver(URI, auth=AUTH)

# MERGE：如果不存在则创建，如果存在则匹配
records, summary, keys = driver.execute_query(
    """
    MERGE (a:Actor {name: $name})
    ON CREATE SET a.born = $born
    ON MATCH SET a.lastSeen = timestamp()
    RETURN a
    """,
    name="周星驰",
    born=1962
)

for record in records:
    print(f"节点：{record['a']}")

driver.close()

4. 使用 Session 方式（可选） #

除了 execute_query()，还可以使用 Session 方式进行更精细的事务控制。

4.1 基本用法 #

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "12345678")
driver = GraphDatabase.driver(URI, auth=AUTH)

def create_actor(tx, name, born):
    """创建演员节点"""
    tx.run("CREATE (a:Actor {name: $name, born: $born})", name=name, born=born)

def find_actor(tx, name):
    """查找演员节点"""
    result = tx.run("MATCH (a:Actor {name: $name}) RETURN a", name=name)
    return [record["a"] for record in result]

with driver.session() as session:
    # 写入操作使用 execute_write
    session.execute_write(create_actor, "吴孟达", 1952)

    # 读取操作使用 execute_read
    actors = session.execute_read(find_actor, "吴孟达")
    print(f"找到的演员：{actors}")

driver.close()

4.2 何时使用 Session 方式？ #

• 需要在同一个事务中执行多个操作
• 需要更精细的事务控制
• 需要区分读写操作以优化性能

5. 处理查询结果 #

5.1 访问记录数据 #

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "12345678")
driver = GraphDatabase.driver(URI, auth=AUTH)

records, summary, keys = driver.execute_query(
    """
    MATCH (a:Actor)
    RETURN a.name AS name, a.born AS born
    LIMIT 5
    """
)

# 方法1：遍历记录列表
print("方法1：遍历记录")
for record in records:
    print(f"  {record['name']}, {record['born']} 年")

# 方法2：转换为字典
if len(records) > 0:
    record_dict = records[0].data()
    print(f"\n记录字典：{record_dict}")

# 方法3：获取所有字段名
print(f"\n结果字段：{keys}")

driver.close()

5.2 处理节点和关系对象 #

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "12345678")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 查询节点
records, summary, keys = driver.execute_query(
    "MATCH (a:Actor {name: $name}) RETURN a",
    name="周星驰"
)

for record in records:
    node = record["a"]
    print(f"节点ID：{node.element_id}")
    print(f"节点标签：{list(node.labels)}")
    print(f"节点属性：{dict(node.items())}")

# 查询关系
records, summary, keys = driver.execute_query(
    """
    MATCH (a:Actor {name: $name})-[r:ACTED_IN]->(m:Movie)
    RETURN a, r, m
    """,
    name="周星驰"
)

for record in records:
    relationship = record["r"]
    print(f"关系类型：{relationship.type}")
    print(f"关系属性：{dict(relationship.items())}")

driver.close()

5.3 查询统计信息 #

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "12345678")
driver = GraphDatabase.driver(URI, auth=AUTH)

records, summary, keys = driver.execute_query(
    "CREATE (a:Actor {name: $name, born: $born}) RETURN a",
    name="赵薇",
    born=1976
)

# 访问统计信息
counters = summary.counters
print("操作统计：")
print(f"  创建的节点数：{counters.nodes_created}")
print(f"  创建的关系数：{counters.relationships_created}")
print(f"  设置的属性数：{counters.properties_set}")
print(f"  删除的节点数：{counters.nodes_deleted}")

# 查询执行时间
print(f"\n查询执行时间：{summary.result_available_after} ms")

driver.close()

6. 错误处理 #

6.1 异常类型 #

Neo4j 驱动定义了多种异常类型：

• AuthError：认证错误（用户名或密码错误）
• ServiceUnavailable：服务不可用（数据库未启动或无法连接）
• ClientError：客户端错误（查询语法错误等）
• Neo4jError：通用 Neo4j 错误

6.2 完整错误处理示例 #

from neo4j import GraphDatabase
from neo4j.exceptions import (
    AuthError,
    ServiceUnavailable,
    ClientError,
    Neo4jError
)

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "12345678")

try:
    driver = GraphDatabase.driver(URI, auth=AUTH)
    driver.verify_connectivity()
    print(" 连接成功！")

    records, summary, keys = driver.execute_query(
        "MATCH (a:Actor) RETURN a LIMIT 10"
    )

    for record in records:
        print(record["a"])

except AuthError as e:
    print(f" 认证失败：用户名或密码错误")

except ServiceUnavailable as e:
    print(f" 服务不可用：无法连接到数据库")
    print(f"   请检查 Neo4j 数据库是否已启动")

except ClientError as e:
    print(f" 查询错误：{e.message}")

except Neo4jError as e:
    print(f" Neo4j 错误：{e.message}")

except Exception as e:
    print(f" 未知错误：{e}")

finally:
    try:
        driver.close()
    except:
        pass

7. 封装为类 #

在实际应用中，建议将数据库操作封装成类：

from neo4j import GraphDatabase

class Neo4jClient:
    """Neo4j 数据库客户端类"""

    def __init__(self, uri, auth):
        self.driver = GraphDatabase.driver(uri, auth=auth)

    def close(self):
        self.driver.close()

    def create_actor(self, name, born):
        """创建演员节点"""
        records, summary, keys = self.driver.execute_query(
            """
            MERGE (a:Actor {name: $name})
            ON CREATE SET a.born = $born
            RETURN a
            """,
            name=name,
            born=born
        )
        return records[0]["a"] if records else None

    def create_movie(self, title, released):
        """创建电影节点"""
        records, summary, keys = self.driver.execute_query(
            """
            MERGE (m:Movie {title: $title})
            ON CREATE SET m.released = $released
            RETURN m
            """,
            title=title,
            released=released
        )
        return records[0]["m"] if records else None

    def create_acted_in(self, actor_name, movie_title, roles):
        """创建出演关系"""
        records, summary, keys = self.driver.execute_query(
            """
            MATCH (a:Actor {name: $actor_name})
            MATCH (m:Movie {title: $movie_title})
            MERGE (a)-[r:ACTED_IN]->(m)
            ON CREATE SET r.roles = $roles
            RETURN r
            """,
            actor_name=actor_name,
            movie_title=movie_title,
            roles=roles
        )
        return records[0]["r"] if records else None

    def find_movies_by_actor(self, actor_name):
        """查找演员出演的电影"""
        records, summary, keys = self.driver.execute_query(
            """
            MATCH (a:Actor {name: $actor_name})-[r:ACTED_IN]->(m:Movie)
            RETURN m.title AS title, r.roles AS roles, m.released AS released
            """,
            actor_name=actor_name
        )
        return [record.data() for record in records]


# 使用示例
if __name__ == "__main__":
    client = Neo4jClient(
        uri="bolt://localhost:7687",
        auth=("neo4j", "12345678")
    )

    try:
        # 创建演员
        client.create_actor("周星驰", 1962)
        client.create_actor("吴孟达", 1952)

        # 创建电影
        client.create_movie("功夫", 2004)
        client.create_movie("少林足球", 2001)

        # 创建出演关系
        client.create_acted_in("周星驰", "功夫", ["阿星"])
        client.create_acted_in("周星驰", "少林足球", ["五师兄"])
        client.create_acted_in("吴孟达", "少林足球", ["黄金右脚"])

        # 查询电影
        movies = client.find_movies_by_actor("周星驰")
        print("周星驰出演的电影：")
        for movie in movies:
            print(f"  《{movie['title']}》({movie['released']}) - 饰演 {movie['roles']}")

    finally:
        client.close()

8. 典型应用场景 #

8.1 电影知识图谱 #

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "12345678")
driver = GraphDatabase.driver(URI, auth=AUTH)

def create_movie_knowledge_graph(driver):
    """创建周星驰电影知识图谱"""
    driver.execute_query("""
        // 创建演员节点
        MERGE (zxc:Actor {name: "周星驰", born: 1962})
        MERGE (wmd:Actor {name: "吴孟达", born: 1952})
        MERGE (zy:Actor {name: "朱茵", born: 1971})
        MERGE (zw:Actor {name: "赵薇", born: 1976})

        // 创建电影节点
        MERGE (kungfu:Movie {title: "功夫", released: 2004})
        MERGE (erta:Movie {title: "大话西游", released: 1995})
        MERGE (shaolin:Movie {title: "少林足球", released: 2001})

        // 创建电影类型节点
        MERGE (action:Genre {name: "动作"})
        MERGE (comedy:Genre {name: "喜剧"})
        MERGE (fantasy:Genre {name: "奇幻"})

        // 创建出演关系
        MERGE (zxc)-[:ACTED_IN {roles: ["阿星"]}]->(kungfu)
        MERGE (zxc)-[:ACTED_IN {roles: ["至尊宝"]}]->(erta)
        MERGE (zxc)-[:ACTED_IN {roles: ["五师兄"]}]->(shaolin)
        MERGE (wmd)-[:ACTED_IN {roles: ["猪八戒"]}]->(erta)
        MERGE (wmd)-[:ACTED_IN {roles: ["黄金右脚"]}]->(shaolin)
        MERGE (zy)-[:ACTED_IN {roles: ["紫霞仙子"]}]->(erta)
        MERGE (zw)-[:ACTED_IN {roles: ["阿梅"]}]->(shaolin)

        // 创建导演关系
        MERGE (zxc)-[:DIRECTED]->(kungfu)
        MERGE (zxc)-[:DIRECTED]->(shaolin)

        // 创建电影类型关系
        MERGE (kungfu)-[:BELONGS_TO]->(action)
        MERGE (kungfu)-[:BELONGS_TO]->(comedy)
        MERGE (shaolin)-[:BELONGS_TO]->(comedy)
        MERGE (erta)-[:BELONGS_TO]->(fantasy)
        MERGE (erta)-[:BELONGS_TO]->(comedy)
    """)
    print(" 电影知识图谱创建成功！")

def find_actors_in_movie(driver, movie_title):
    """查找电影中的演员"""
    records, _, _ = driver.execute_query(
        """
        MATCH (m:Movie {title: $title})<-[r:ACTED_IN]-(a:Actor)
        RETURN a.name AS actor, r.roles AS roles
        """,
        title=movie_title
    )
    return [{"actor": r["actor"], "roles": r["roles"]} for r in records]

def find_movies_by_genre(driver, genre_name):
    """查找特定类型的电影"""
    records, _, _ = driver.execute_query(
        """
        MATCH (g:Genre {name: $genre})<-[:BELONGS_TO]-(m:Movie)
        RETURN m.title AS title, m.released AS released
        """,
        genre=genre_name
    )
    return [{"title": r["title"], "released": r["released"]} for r in records]

# 使用示例
create_movie_knowledge_graph(driver)

actors = find_actors_in_movie(driver, "少林足球")
print("\n《少林足球》的演员：")
for a in actors:
    print(f"  {a['actor']} - 饰演 {a['roles']}")

movies = find_movies_by_genre(driver, "喜剧")
print("\n喜剧类型的电影：")
for m in movies:
    print(f"  《{m['title']}》({m['released']})")

driver.close()

8.2 社交网络好友推荐 #

from neo4j import GraphDatabase

class SocialNetwork:
    """社交网络应用"""

    def __init__(self, uri, auth):
        self.driver = GraphDatabase.driver(uri, auth=auth)

    def close(self):
        self.driver.close()

    def add_user(self, user_id, name):
        """添加用户"""
        self.driver.execute_query(
            "MERGE (u:User {id: $id}) SET u.name = $name",
            id=user_id, name=name
        )

    def add_friendship(self, user1_id, user2_id):
        """添加朋友关系"""
        self.driver.execute_query(
            """
            MATCH (u1:User {id: $id1}), (u2:User {id: $id2})
            MERGE (u1)-[:FRIEND]-(u2)
            """,
            id1=user1_id, id2=user2_id
        )

    def recommend_friends(self, user_id, limit=5):
        """推荐朋友（朋友的朋友）"""
        records, _, _ = self.driver.execute_query(
            """
            MATCH (user:User {id: $id})-[:FRIEND]-(friend)-[:FRIEND]-(potential:User)
            WHERE user <> potential AND NOT (user)-[:FRIEND]-(potential)
            WITH potential, COUNT(DISTINCT friend) AS common_friends
            RETURN potential.name AS name, common_friends
            ORDER BY common_friends DESC
            LIMIT $limit
            """,
            id=user_id, limit=limit
        )
        return [{"name": r["name"], "common_friends": r["common_friends"]} for r in records]

# 使用示例
if __name__ == "__main__":
    social = SocialNetwork("bolt://localhost:7687", ("neo4j", "12345678"))

    try:
        # 添加用户
        for i, name in enumerate(["张三", "李四", "王五", "赵六", "钱七"], 1):
            social.add_user(f"u{i}", name)

        # 添加朋友关系
        social.add_friendship("u1", "u2")  # 张三 <-> 李四
        social.add_friendship("u1", "u3")  # 张三 <-> 王五
        social.add_friendship("u2", "u4")  # 李四 <-> 赵六
        social.add_friendship("u3", "u4")  # 王五 <-> 赵六
        social.add_friendship("u4", "u5")  # 赵六 <-> 钱七

        # 为张三推荐朋友
        recommendations = social.recommend_friends("u1")
        print("为张三推荐的朋友：")
        for rec in recommendations:
            print(f"  {rec['name']} (共同朋友数：{rec['common_friends']})")

    finally:
        social.close()

9. 最佳实践 #

9.1 使用参数化查询 #

推荐：使用参数

driver.execute_query(
    "MATCH (a:Actor {name: $name}) RETURN a",
    name="周星驰"
)

不推荐：字符串拼接

name = "周星驰"
driver.execute_query(f"MATCH (a:Actor {{name: '{name}'}}) RETURN a")

9.2 使用上下文管理器 #

from neo4j import GraphDatabase
from contextlib import contextmanager

@contextmanager
def get_driver(uri, auth):
    """创建数据库驱动的上下文管理器"""
    driver = None
    try:
        driver = GraphDatabase.driver(uri, auth=auth)
        driver.verify_connectivity()
        yield driver
    finally:
        if driver:
            driver.close()

# 使用示例
with get_driver("bolt://localhost:7687", ("neo4j", "12345678")) as driver:
    records, _, _ = driver.execute_query("MATCH (a:Actor) RETURN a.name LIMIT 5")
    for record in records:
        print(record["a.name"])

9.3 总是关闭连接 #

driver = GraphDatabase.driver(URI, auth=AUTH)
try:
    # 执行操作
    pass
finally:
    driver.close()

9.4 处理异常 #

try:
    driver.verify_connectivity()
except Exception as e:
    print(f" 连接失败：{e}")

10. 小结 #

10.1 核心概念 #

概念	说明	主要用途
GraphDatabase.driver	创建数据库驱动	连接 Neo4j 数据库
execute_query()	执行查询方法	最简单的查询方式
session.execute_write()	执行写入操作	创建、更新、删除数据
session.execute_read()	执行读取操作	查询数据
MERGE	创建或匹配数据	避免重复数据

10.2 最佳实践 #

1. 使用参数化查询：安全且高效
2. 使用 execute_query()：简单场景首选
3. 使用 Session 方式：需要事务控制时使用
4. 总是关闭连接：释放资源
5. 处理异常：提高程序健壮性
6. 封装为类：方便管理和复用

10.3 常用操作速查 #

# 创建连接
driver = GraphDatabase.driver(URI, auth=AUTH)

# 执行查询
records, summary, keys = driver.execute_query("MATCH (n) RETURN n LIMIT 10")

# 参数化查询
records, _, _ = driver.execute_query("MATCH (n {name: $name}) RETURN n", name="周星驰")

# 关闭连接
driver.close()