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 等）

如果你还不熟悉这些，建议先学习 Neo4j 和 Cypher 的基础知识。

1.3 安装驱动 #

在开始之前，我们需要先安装 neo4j 驱动。

安装步骤：

# 使用 pip 安装 neo4j 驱动
# pip 是 Python 的包管理工具
pip install neo4j

验证安装：

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

2. 连接数据库 #

2.1 基础连接 #

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

连接信息说明：

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

完整示例：

# 导入 Neo4j 驱动
# GraphDatabase 是创建数据库连接的入口类
from neo4j import GraphDatabase

# 数据库连接信息
# URI 格式：neo4j://主机地址:端口号
# localhost 表示本地数据库
# 7687 是 Neo4j 的默认 Bolt 端口
URI = "neo4j://localhost:7687"

# 认证信息
# 格式：(用户名, 密码)
# 默认用户名通常是 "neo4j"
AUTH = ("neo4j", "your_password")

# 创建数据库驱动
# driver() 方法创建并返回一个驱动对象
# 驱动对象用于管理数据库连接
driver = GraphDatabase.driver(URI, auth=AUTH)

# 验证连接
# verify_connectivity() 方法测试连接是否成功
# 如果连接失败会抛出异常
try:
    driver.verify_connectivity()
    print("✅ 连接成功！")
except Exception as e:
    print(f"❌ 连接失败：{e}")

# 关闭连接
# 使用完驱动后，记得关闭连接释放资源
driver.close()

2.2 连接配置 #

在实际应用中，我们可能需要配置一些连接参数，如连接池大小、超时时间等。

常用配置参数：

• max_connection_pool_size：连接池最大连接数
• connection_timeout：连接超时时间（秒）
• encrypted：是否启用加密（生产环境建议启用）

完整示例：

# 导入 Neo4j 驱动
from neo4j import GraphDatabase

# 数据库连接信息
URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your_password")

# 连接配置
# 使用字典配置连接参数
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 驱动提供的最简单、最常用的查询方法。它会自动管理会话和事务，非常适合初学者使用。

方法特点：

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

3.2 基础查询 #

让我们从最简单的查询开始，学习如何使用 execute_query() 方法。

完整示例：

# 导入 Neo4j 驱动
from neo4j import GraphDatabase

# 创建数据库连接
URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your_password")
driver = GraphDatabase.driver(URI, auth=AUTH)

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

# 处理查询结果
# records 是一个记录列表，每个记录包含查询返回的数据
for record in records:
    # 通过键名访问记录的值
    print(record["message"])

# 关闭连接
driver.close()

3.3 参数化查询 #

在实际应用中，我们经常需要根据不同的条件查询数据。使用参数化查询可以避免 SQL 注入，提高安全性。

参数化查询的优势：

• 安全性：防止注入攻击
• 性能：查询计划可以复用
• 可读性：代码更清晰

完整示例：

# 导入 Neo4j 驱动
from neo4j import GraphDatabase

# 创建数据库连接
URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your_password")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 参数化查询
# 在 Cypher 查询中使用 $参数名 的形式定义参数
# 在 execute_query() 中通过关键字参数传递参数值
records, summary, keys = driver.execute_query(
    "MATCH (p:Person) WHERE p.age > $min_age RETURN p.name AS name, p.age AS age",
    min_age=25  # 传递参数值
)

# 处理查询结果
print("年龄大于 25 的人员：")
for record in records:
    # 访问记录中的字段
    name = record["name"]
    age = record["age"]
    print(f"  {name}: {age} 岁")

# 关闭连接
driver.close()

3.4 创建数据 #

使用 execute_query() 可以轻松创建节点和关系。

创建节点示例：

# 导入 Neo4j 驱动
from neo4j import GraphDatabase

# 创建数据库连接
URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your_password")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 创建节点
# CREATE 语句用于创建新节点
# 使用参数化查询传递节点属性
records, summary, keys = driver.execute_query(
    """
    CREATE (p:Person {name: $name, age: $age, city: $city})
    RETURN p
    """,
    name="张三",
    age=25,
    city="北京"
)

# 处理结果
for record in records:
    # record["p"] 是创建的节点对象
    person = record["p"]
    print(f"创建节点：{person}")

# 查看创建统计
# summary.counters 包含操作统计信息
print(f"创建的节点数：{summary.counters.nodes_created}")

# 关闭连接
driver.close()

创建关系示例：

# 导入 Neo4j 驱动
from neo4j import GraphDatabase

# 创建数据库连接
URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your_password")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 创建关系
# 首先匹配两个节点，然后创建它们之间的关系
records, summary, keys = driver.execute_query(
    """
    MATCH (a:Person {name: $name1})
    MATCH (b:Person {name: $name2})
    CREATE (a)-[r:KNOWS {since: $since}]->(b)
    RETURN r
    """,
    name1="张三",
    name2="李四",
    since=2020
)

# 处理结果
for record in records:
    relationship = record["r"]
    print(f"创建关系：{relationship}")

# 查看创建统计
print(f"创建的关系数：{summary.counters.relationships_created}")

# 关闭连接
driver.close()

3.5 更新数据 #

使用 SET 语句可以更新节点的属性。

完整示例：

# 导入 Neo4j 驱动
from neo4j import GraphDatabase

# 创建数据库连接
URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your_password")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 更新节点属性
# MATCH 匹配节点，SET 更新属性
records, summary, keys = driver.execute_query(
    """
    MATCH (p:Person {name: $name})
    SET p.age = $new_age, p.city = $new_city
    RETURN p
    """,
    name="张三",
    new_age=26,
    new_city="上海"
)

# 处理结果
for record in records:
    person = record["p"]
    print(f"更新后的节点：{person}")

# 查看更新统计
print(f"更新的节点数：{summary.counters.properties_set}")

# 关闭连接
driver.close()

3.6 删除数据 #

使用 DELETE 或 DETACH DELETE 可以删除节点和关系。

删除关系示例：

# 导入 Neo4j 驱动
from neo4j import GraphDatabase

# 创建数据库连接
URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your_password")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 删除关系
# DELETE 只删除关系，不删除节点
records, summary, keys = driver.execute_query(
    """
    MATCH (a:Person {name: $name1})-[r:KNOWS]-(b:Person {name: $name2})
    DELETE r
    RETURN count(r) AS deleted_count
    """,
    name1="张三",
    name2="李四"
)

# 处理结果
for record in records:
    print(f"删除的关系数：{record['deleted_count']}")

# 关闭连接
driver.close()

删除节点示例：

# 导入 Neo4j 驱动
from neo4j import GraphDatabase

# 创建数据库连接
URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your_password")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 删除节点
# DETACH DELETE 会删除节点及其所有关系
records, summary, keys = driver.execute_query(
    """
    MATCH (p:Person {name: $name})
    DETACH DELETE p
    RETURN count(p) AS deleted_count
    """,
    name="张三"
)

# 处理结果
for record in records:
    print(f"删除的节点数：{record['deleted_count']}")

# 关闭连接
driver.close()

4. 处理查询结果 #

4.1 前置知识：查询结果结构 #

execute_query() 方法返回三个值：

• records：记录列表，包含查询返回的所有数据
• summary：查询摘要，包含统计信息和元数据
• keys：结果键列表，表示返回的字段名

4.2 访问记录数据 #

记录（Record）是查询结果的基本单位，每个记录包含一行数据。

完整示例：

# 导入 Neo4j 驱动
from neo4j import GraphDatabase

# 创建数据库连接
URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your_password")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 执行查询
records, summary, keys = driver.execute_query(
    """
    MATCH (p:Person)
    RETURN p.name AS name, p.age AS age, p.city AS city
    LIMIT 5
    """
)

# 方法1：遍历记录列表
# records 是一个列表，可以直接遍历
print("方法1：遍历记录列表")
for record in records:
    # 通过键名访问字段值
    name = record["name"]
    age = record["age"]
    city = record["city"]
    print(f"  {name}, {age} 岁, {city}")

# 方法2：访问单个记录
# 如果查询只返回一条记录，可以使用索引访问
if len(records) > 0:
    first_record = records[0]
    print(f"\n第一条记录：{first_record['name']}")

# 方法3：获取所有字段
# 使用 keys() 方法获取记录的所有字段名
if len(records) > 0:
    print(f"\n记录字段：{list(records[0].keys())}")

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

# 关闭连接
driver.close()

4.3 处理节点和关系对象 #

当查询返回节点或关系对象时，我们可以访问它们的属性。

完整示例：

# 导入 Neo4j 驱动
from neo4j import GraphDatabase

# 创建数据库连接
URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your_password")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 查询节点
records, summary, keys = driver.execute_query(
    """
    MATCH (p:Person {name: $name})
    RETURN p
    """,
    name="张三"
)

# 处理节点对象
for record in records:
    # record["p"] 是节点对象
    node = record["p"]

    # 访问节点属性
    # 节点对象可以像字典一样访问属性
    print(f"节点ID：{node.element_id}")
    print(f"节点标签：{list(node.labels)}")
    print(f"节点属性：{dict(node.items())}")

    # 访问单个属性
    if "name" in node:
        print(f"姓名：{node['name']}")

# 查询关系
records, summary, keys = driver.execute_query(
    """
    MATCH (a:Person {name: $name1})-[r:KNOWS]-(b:Person {name: $name2})
    RETURN r, a, b
    """,
    name1="张三",
    name2="李四"
)

# 处理关系对象
for record in records:
    # record["r"] 是关系对象
    relationship = record["r"]

    # 访问关系属性
    print(f"关系类型：{relationship.type}")
    print(f"关系属性：{dict(relationship.items())}")

    # 访问起始节点和结束节点
    start_node = record["a"]
    end_node = record["b"]
    print(f"起始节点：{start_node['name']}")
    print(f"结束节点：{end_node['name']}")

# 关闭连接
driver.close()

4.4 查询统计信息 #

summary 对象包含查询执行的统计信息，如影响的节点数、关系数等。

完整示例：

# 导入 Neo4j 驱动
from neo4j import GraphDatabase

# 创建数据库连接
URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your_password")
driver = GraphDatabase.driver(URI, auth=AUTH)

# 执行创建操作
records, summary, keys = driver.execute_query(
    """
    CREATE (p:Person {name: $name, age: $age})
    RETURN p
    """,
    name="王五",
    age=30
)

# 访问统计信息
# summary.counters 包含操作计数器
counters = summary.counters

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

# 查询执行时间
# result_available_after 表示结果可用的时间（毫秒）
# result_consumed_after 表示结果消耗的时间（毫秒）
print(f"\n查询执行时间：{summary.result_available_after} ms")

# 关闭连接
driver.close()

5. 封装为类 #

5.1 创建数据库操作类 #

在实际应用中，我们通常会将数据库操作封装成一个类，方便管理和复用。

完整示例：

# 导入 Neo4j 驱动
from neo4j import GraphDatabase

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

    def __init__(self, uri, auth):
        """
        初始化客户端

        参数：
        - uri: 数据库 URI
        - auth: 认证信息 (用户名, 密码)
        """
        # 创建数据库驱动
        self.driver = GraphDatabase.driver(uri, auth=auth)

    def close(self):
        """关闭数据库连接"""
        # 关闭驱动，释放资源
        self.driver.close()

    def create_person(self, name, age, city):
        """
        创建人员节点

        参数：
        - name: 姓名
        - age: 年龄
        - city: 城市

        返回：
        - 创建的节点
        """
        # 执行创建查询
        records, summary, keys = self.driver.execute_query(
            """
            CREATE (p:Person {name: $name, age: $age, city: $city})
            RETURN p
            """,
            name=name,
            age=age,
            city=city
        )

        # 返回创建的节点
        if records:
            return records[0]["p"]
        return None

    def find_person(self, name):
        """
        查找人员节点

        参数：
        - name: 姓名

        返回：
        - 节点对象，如果不存在返回 None
        """
        # 执行查询
        records, summary, keys = self.driver.execute_query(
            """
            MATCH (p:Person {name: $name})
            RETURN p
            """,
            name=name
        )

        # 返回第一个匹配的节点
        if records:
            return records[0]["p"]
        return None

    def create_friendship(self, name1, name2, since=None):
        """
        创建朋友关系

        参数：
        - name1: 第一个人的姓名
        - name2: 第二个人的姓名
        - since: 成为朋友的年份（可选）

        返回：
        - 创建的关系
        """
        # 构建查询
        if since:
            # 如果有年份，添加关系属性
            query = """
            MATCH (a:Person {name: $name1})
            MATCH (b:Person {name: $name2})
            MERGE (a)-[r:KNOWS {since: $since}]->(b)
            RETURN r
            """
            records, summary, keys = self.driver.execute_query(
                query,
                name1=name1,
                name2=name2,
                since=since
            )
        else:
            # 如果没有年份，不添加属性
            query = """
            MATCH (a:Person {name: $name1})
            MATCH (b:Person {name: $name2})
            MERGE (a)-[r:KNOWS]->(b)
            RETURN r
            """
            records, summary, keys = self.driver.execute_query(
                query,
                name1=name1,
                name2=name2
            )

        # 返回创建的关系
        if records:
            return records[0]["r"]
        return None

    def get_friends(self, name):
        """
        获取某个人的所有朋友

        参数：
        - name: 姓名

        返回：
        - 朋友列表
        """
        # 执行查询
        records, summary, keys = self.driver.execute_query(
            """
            MATCH (p:Person {name: $name})-[:KNOWS]-(friend:Person)
            RETURN friend.name AS name, friend.age AS age
            """,
            name=name
        )

        # 转换为列表
        friends = []
        for record in records:
            friends.append({
                "name": record["name"],
                "age": record["age"]
            })

        return friends


# 使用示例
if __name__ == "__main__":
    # 创建客户端
    client = Neo4jClient(
        uri="neo4j://localhost:7687",
        auth=("neo4j", "your_password")
    )

    try:
        # 创建人员
        person1 = client.create_person("张三", 25, "北京")
        person2 = client.create_person("李四", 30, "上海")

        # 创建朋友关系
        friendship = client.create_friendship("张三", "李四", since=2020)

        # 查找人员
        found_person = client.find_person("张三")
        if found_person:
            print(f"找到人员：{found_person['name']}")

        # 获取朋友列表
        friends = client.get_friends("张三")
        print(f"张三的朋友：{friends}")

    finally:
        # 关闭连接
        client.close()

6. 错误处理 #

6.1 前置知识：异常类型 #

Neo4j 驱动定义了多种异常类型，用于处理不同的错误情况：

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

6.2 基础错误处理 #

在实际应用中，我们需要处理可能出现的错误，提供友好的错误提示。

完整示例：

# 导入 Neo4j 驱动和异常类
from neo4j import GraphDatabase
from neo4j.exceptions import (
    AuthError,
    ServiceUnavailable,
    ClientError,
    Neo4jError
)

# 创建数据库连接
URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your_password")

try:
    # 创建驱动
    driver = GraphDatabase.driver(URI, auth=AUTH)

    # 验证连接
    driver.verify_connectivity()
    print("✅ 连接成功！")

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

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

except AuthError as e:
    # 处理认证错误
    print(f"❌ 认证失败：用户名或密码错误")
    print(f"   错误信息：{e}")

except ServiceUnavailable as e:
    # 处理服务不可用错误
    print(f"❌ 服务不可用：无法连接到数据库")
    print(f"   请检查：")
    print(f"   1. Neo4j 数据库是否已启动")
    print(f"   2. URI 地址是否正确")
    print(f"   3. 网络连接是否正常")
    print(f"   错误信息：{e}")

except ClientError as e:
    # 处理客户端错误（通常是查询语法错误）
    print(f"❌ 查询错误：{e.message}")
    print(f"   错误代码：{e.code}")

except Neo4jError as e:
    # 处理其他 Neo4j 错误
    print(f"❌ Neo4j 错误：{e.message}")
    print(f"   错误代码：{e.code}")

except Exception as e:
    # 处理其他未知错误
    print(f"❌ 未知错误：{e}")

finally:
    # 确保关闭连接
    try:
        driver.close()
    except:
        pass

6.3 健壮的操作函数 #

在实际应用中，我们可以创建一个健壮的操作函数，自动处理错误。

完整示例：

# 导入必要的库
from neo4j import GraphDatabase
from neo4j.exceptions import Neo4jError

def safe_execute_query(driver, query, **parameters):
    """
    安全执行查询，自动处理错误

    参数：
    - driver: 数据库驱动
    - query: Cypher 查询语句
    - **parameters: 查询参数

    返回：
    - (records, summary, keys) 如果成功
    - None 如果失败
    """
    try:
        # 执行查询
        records, summary, keys = driver.execute_query(query, **parameters)
        return records, summary, keys

    except Neo4jError as e:
        # 打印错误信息
        print(f"❌ 查询执行失败：{e.message}")
        return None, None, None

    except Exception as e:
        # 处理其他错误
        print(f"❌ 未知错误：{e}")
        return None, None, None


# 使用示例
if __name__ == "__main__":
    # 创建驱动
    driver = GraphDatabase.driver(
        "neo4j://localhost:7687",
        auth=("neo4j", "your_password")
    )

    try:
        # 使用安全查询函数
        records, summary, keys = safe_execute_query(
            driver,
            "MATCH (p:Person) RETURN p.name AS name LIMIT 5"
        )

        # 检查是否成功
        if records is not None:
            print("查询成功！")
            for record in records:
                print(f"  {record['name']}")
        else:
            print("查询失败！")

    finally:
        # 关闭连接
        driver.close()

7. 实际应用示例 #

7.1 社交网络应用 #

让我们创建一个完整的社交网络应用示例，展示如何使用 Neo4j 驱动实现好友推荐功能。

完整示例：

# 导入必要的库
from neo4j import GraphDatabase

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

    def __init__(self, uri, auth):
        """
        初始化社交网络应用

        参数：
        - uri: 数据库 URI
        - auth: 认证信息
        """
        # 创建数据库驱动
        self.driver = GraphDatabase.driver(uri, auth=auth)

    def close(self):
        """关闭数据库连接"""
        self.driver.close()

    def add_user(self, user_id, name, age):
        """
        添加用户

        参数：
        - user_id: 用户ID
        - name: 姓名
        - age: 年龄
        """
        # 创建用户节点
        records, summary, keys = self.driver.execute_query(
            """
            MERGE (u:User {id: $user_id})
            SET u.name = $name, u.age = $age
            RETURN u
            """,
            user_id=user_id,
            name=name,
            age=age
        )

        if records:
            print(f"✅ 添加用户：{records[0]['u']['name']}")

    def add_friendship(self, user1_id, user2_id):
        """
        添加朋友关系

        参数：
        - user1_id: 第一个用户ID
        - user2_id: 第二个用户ID
        """
        # 创建双向朋友关系（无向图）
        records, summary, keys = self.driver.execute_query(
            """
            MATCH (u1:User {id: $id1})
            MATCH (u2:User {id: $id2})
            MERGE (u1)-[:FRIEND]-(u2)
            RETURN u1, u2
            """,
            id1=user1_id,
            id2=user2_id
        )

        if records:
            print(f"✅ 添加朋友关系：{records[0]['u1']['name']} <-> {records[0]['u2']['name']}")

    def recommend_friends(self, user_id, limit=5):
        """
        推荐朋友（朋友的朋友）

        参数：
        - user_id: 用户ID
        - limit: 推荐数量

        返回：
        - 推荐的朋友列表
        """
        # 查找朋友的朋友（2度关系）
        # 排除已经是朋友的人
        records, summary, keys = self.driver.execute_query(
            """
            MATCH (user:User {id: $user_id})-[:FRIEND]-(friend)-[:FRIEND]-(potential:User)
            WHERE user <> potential
            AND NOT (user)-[:FRIEND]-(potential)
            WITH potential, COUNT(DISTINCT friend) AS common_friends
            RETURN potential.id AS id, potential.name AS name, common_friends
            ORDER BY common_friends DESC
            LIMIT $limit
            """,
            user_id=user_id,
            limit=limit
        )

        # 转换为列表
        recommendations = []
        for record in records:
            recommendations.append({
                "id": record["id"],
                "name": record["name"],
                "common_friends": record["common_friends"]
            })

        return recommendations

    def get_friends(self, user_id):
        """
        获取用户的所有朋友

        参数：
        - user_id: 用户ID

        返回：
        - 朋友列表
        """
        # 查询所有朋友
        records, summary, keys = self.driver.execute_query(
            """
            MATCH (user:User {id: $user_id})-[:FRIEND]-(friend:User)
            RETURN friend.id AS id, friend.name AS name, friend.age AS age
            """,
            user_id=user_id
        )

        # 转换为列表
        friends = []
        for record in records:
            friends.append({
                "id": record["id"],
                "name": record["name"],
                "age": record["age"]
            })

        return friends


# 使用示例
if __name__ == "__main__":
    # 创建社交网络应用
    social = SocialNetwork(
        uri="neo4j://localhost:7687",
        auth=("neo4j", "your_password")
    )

    try:
        # 添加用户
        social.add_user("u1", "张三", 25)
        social.add_user("u2", "李四", 30)
        social.add_user("u3", "王五", 28)
        social.add_user("u4", "赵六", 32)
        social.add_user("u5", "钱七", 26)

        # 添加朋友关系
        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")  # 赵六 <-> 钱七

        # 获取朋友列表
        print("\n张三的朋友：")
        friends = social.get_friends("u1")
        for friend in friends:
            print(f"  {friend['name']}")

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

    finally:
        # 关闭连接
        social.close()

8. 最佳实践 #

8.1 使用上下文管理器 #

使用上下文管理器可以确保连接总是被正确关闭，即使发生错误也不例外。

完整示例：

# 导入必要的库
from neo4j import GraphDatabase
from contextlib import contextmanager

@contextmanager
def get_driver(uri, auth):
    """
    创建数据库驱动的上下文管理器

    参数：
    - uri: 数据库 URI
    - auth: 认证信息

    使用：
    with get_driver(uri, auth) as driver:
        # 使用 driver
        pass
    """
    # 创建驱动
    driver = None
    try:
        driver = GraphDatabase.driver(uri, auth=auth)
        # 验证连接
        driver.verify_connectivity()
        # 返回驱动
        yield driver
    finally:
        # 确保关闭连接
        if driver:
            driver.close()


# 使用示例
if __name__ == "__main__":
    URI = "neo4j://localhost:7687"
    AUTH = ("neo4j", "your_password")

    # 使用上下文管理器
    # 即使发生错误，连接也会自动关闭
    with get_driver(URI, AUTH) as driver:
        # 执行查询
        records, summary, keys = driver.execute_query(
            "MATCH (p:Person) RETURN p.name AS name LIMIT 5"
        )

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

8.2 参数化查询 #

始终使用参数化查询，避免字符串拼接，提高安全性和性能。

好的做法：

# 使用参数化查询（推荐）
records, summary, keys = driver.execute_query(
    "MATCH (p:Person {name: $name}) RETURN p",
    name="张三"
)

不好的做法：

# 字符串拼接（不推荐，有安全风险）
name = "张三"
query = f"MATCH (p:Person {{name: '{name}'}}) RETURN p"
records, summary, keys = driver.execute_query(query)

9. 小结 #

9.1 核心概念回顾 #

• 驱动（Driver）：管理数据库连接的对象
• execute_query()：最简单、最常用的查询方法
• 参数化查询：使用 $参数名 的形式传递参数
• 记录（Record）：查询结果的基本单位
• 错误处理：使用 try-except 处理异常

9.2 关键要点 #

1. 使用 execute_query() 方法最简单，适合大多数场景
2. 始终使用参数化查询，提高安全性和性能
3. 记得关闭驱动连接，释放资源
4. 使用上下文管理器可以自动管理连接生命周期
5. 将数据库操作封装成类，方便管理和复用

9.3 下一步学习 #

掌握了 Neo4j Python 驱动的基础后，你可以：

• 学习更复杂的 Cypher 查询
• 学习批量操作和性能优化
• 学习事务管理（显式事务）
• 探索 Neo4j 的高级功能

---

恭喜！ 你已经掌握了 Neo4j Python 驱动的基础知识。现在可以开始创建自己的图数据库应用了！