1. 前置知识 #

在Python中使用Neo4j，需要使用Neo4j官方提供的Python驱动。这个驱动允许你在Python程序中执行Cypher查询，操作图数据。

1.1 什么是数据库驱动？ #

数据库驱动（Driver）是一个程序库，它让你的编程语言（比如 Python）能够和数据库（比如 Neo4j）进行通信。就像翻译官一样，它把 Python 的指令翻译成数据库能理解的语言。

1.2 为什么需要Python驱动？ #

虽然Neo4j提供了浏览器界面可以直接执行Cypher查询，但在实际应用中，我们通常需要：

• 在Python程序中操作数据库
• 将Python程序与Neo4j数据库集成
• 自动化数据处理流程

2. 安装和连接 #

2.1 安装Python驱动 #

在Python中使用Neo4j，需要安装neo4j驱动包。

# 使用 pip 安装
pip install neo4j

2.2 连接Neo4j数据库 #

下面是一个完整的连接示例，展示了如何用Python连接到Neo4j数据库。

# 导入 Neo4j 驱动库
from neo4j import GraphDatabase

# 定义 Neo4j 数据库的连接地址
URI = "bolt://localhost:7687"
# 定义用户名
USERNAME = "neo4j"
# 定义密码（替换为你自己的密码）
PASSWORD = "your_password"

# 创建数据库驱动对象，用于连接 Neo4j
driver = GraphDatabase.driver(URI, auth=(USERNAME, PASSWORD))

# 尝试连接数据库，如果连接失败会捕获异常
try:
    # 验证与 Neo4j 数据库的连接
    driver.verify_connectivity()
    # 如果连接成功，打印提示信息
    print("成功连接到 Neo4j 数据库！")
except Exception as e:
    # 如果连接失败，输出失败信息和异常原因
    print(f"❌ 连接失败：{e}")

# 关闭数据库连接，释放资源
driver.close()

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

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

3. 基本操作 #

3.1 创建节点和关系 #

使用session.execute_write()执行写入操作。

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
USERNAME = "neo4j"
PASSWORD = "your_password"

driver = GraphDatabase.driver(URI, auth=(USERNAME, PASSWORD))

def create_person(tx, name, age):
    """创建Person节点"""
    tx.run("CREATE (p:Person {name: $name, age: $age})", name=name, age=age)

def create_friendship(tx, person1_name, person2_name):
    """创建朋友关系"""
    tx.run("""
        MATCH (p1:Person {name: $person1_name})
        MATCH (p2:Person {name: $person2_name})
        CREATE (p1)-[:IS_FRIENDS_WITH]->(p2)
    """, person1_name=person1_name, person2_name=person2_name)

with driver.session() as session:
    # 创建节点
    session.execute_write(create_person, "张三", 30)
    session.execute_write(create_person, "李四", 25)

    # 创建关系
    session.execute_write(create_friendship, "张三", "李四")
    print("节点和关系创建成功！")

driver.close()

3.2 查询数据 #

使用session.execute_read()执行读取操作。

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
USERNAME = "neo4j"
PASSWORD = "your_password"

driver = GraphDatabase.driver(URI, auth=(USERNAME, PASSWORD))

def find_person(tx, name):
    """查找Person节点"""
    result = tx.run("MATCH (p:Person {name: $name}) RETURN p", name=name)
    return [record["p"] for record in result]

def find_friends(tx, person_name):
    """查找某个人的所有朋友"""
    result = tx.run("""
        MATCH (p:Person {name: $person_name})-[:IS_FRIENDS_WITH]->(friend)
        RETURN friend.name AS name
    """, person_name=person_name)
    return [record["name"] for record in result]

with driver.session() as session:
    # 查找节点
    persons = session.execute_read(find_person, "张三")
    print(f"找到的节点：{persons}")

    # 查找朋友
    friends = session.execute_read(find_friends, "张三")
    print(f"张三的朋友：{friends}")

driver.close()

3.3 使用MERGE避免重复 #

使用MERGE可以避免创建重复数据。

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
USERNAME = "neo4j"
PASSWORD = "your_password"

driver = GraphDatabase.driver(URI, auth=(USERNAME, PASSWORD))

def create_or_get_person(tx, name, age):
    """创建或获取Person节点（使用MERGE）"""
    result = tx.run("""
        MERGE (p:Person {name: $name})
        ON CREATE SET p.age = $age
        ON MATCH SET p.age = $age
        RETURN p
    """, name=name, age=age)
    return result.single()["p"]

with driver.session() as session:
    # 第一次执行：创建节点
    person1 = session.execute_write(create_or_get_person, "张三", 30)
    print(f"创建/获取节点：{person1}")

    # 第二次执行：获取已存在的节点（不会重复创建）
    person2 = session.execute_write(create_or_get_person, "张三", 30)
    print(f"创建/获取节点：{person2}")

driver.close()

4. 典型应用场景 #

4.1 社交网络分析 #

社交网络分析

场景描述： 分析用户之间的朋友关系，推荐可能认识的人。

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
USERNAME = "neo4j"
PASSWORD = "your_password"

driver = GraphDatabase.driver(URI, auth=(USERNAME, PASSWORD))

def create_friendship(tx, person1_name, person2_name):
    """创建朋友关系"""
    tx.run("""
        MERGE (p1:Person {name: $person1_name})
        MERGE (p2:Person {name: $person2_name})
        MERGE (p1)-[:IS_FRIENDS_WITH]->(p2)
    """, person1_name=person1_name, person2_name=person2_name)

def find_friends(tx, person_name):
    """查找某个人的所有朋友"""
    result = tx.run("""
        MATCH (p:Person {name: $person_name})-[:IS_FRIENDS_WITH]->(friend)
        RETURN friend.name AS name
    """, person_name=person_name)
    return [record["name"] for record in result]

def find_friends_of_friends(tx, person_name):
    """查找某个人的朋友的朋友"""
    result = tx.run("""
        MATCH (me:Person {name: $person_name})-[:IS_FRIENDS_WITH*2]->(friend_of_friend)
        RETURN DISTINCT friend_of_friend.name AS name
    """, person_name=person_name)
    return [record["name"] for record in result]

with driver.session() as session:
    # 创建朋友关系
    session.execute_write(create_friendship, "张三", "李四")
    session.execute_write(create_friendship, "张三", "王五")
    session.execute_write(create_friendship, "李四", "王五")

    print("朋友关系创建成功！")

    # 查询朋友
    friends = session.execute_read(find_friends, "张三")
    print(f"张三的朋友：{friends}")

    # 查询朋友的朋友
    friends_of_friends = session.execute_read(find_friends_of_friends, "张三")
    print(f"张三的朋友的朋友：{friends_of_friends}")

driver.close()

4.2 推荐系统 #

推荐系统

场景描述： 基于用户的行为和偏好推荐商品。

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
USERNAME = "neo4j"
PASSWORD = "your_password"

driver = GraphDatabase.driver(URI, auth=(USERNAME, PASSWORD))

def create_data(tx):
    """创建用户、商品和喜欢关系"""
    tx.run("""
        MERGE (u1:User {name: "Alice"})
        MERGE (u2:User {name: "Bob"})
        MERGE (p1:Product {name: "书包"})
        MERGE (p2:Product {name: "水杯"})
        MERGE (p3:Product {name: "运动鞋"})
        MERGE (u1)-[:LIKES]->(p1)
        MERGE (u1)-[:LIKES]->(p2)
        MERGE (u2)-[:LIKES]->(p2)
        MERGE (u2)-[:LIKES]->(p3)
    """)

def recommend_by_similarity(tx, username):
    """基于相似用户推荐商品"""
    result = tx.run("""
        MATCH (me:User {name: $username})-[:LIKES]->(p:Product)<-[:LIKES]-(other:User)-[:LIKES]->(rec:Product)
        WHERE NOT (me)-[:LIKES]->(rec)
        RETURN DISTINCT rec.name AS recommendation
    """, username=username)
    return [record["recommendation"] for record in result]

with driver.session() as session:
    # 创建测试数据
    session.execute_write(create_data)

    # 为Alice推荐商品
    recommendations = session.execute_read(recommend_by_similarity, "Alice")
    print(f"为 Alice 推荐的商品：{recommendations}")

driver.close()

4.3 知识图谱 #

知识图谱

场景描述： 组织和管理复杂的知识关系。

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
USERNAME = "neo4j"
PASSWORD = "your_password"

driver = GraphDatabase.driver(URI, auth=(USERNAME, PASSWORD))

def create_movie_knowledge_graph(tx):
    """创建电影知识图谱"""
    tx.run("""
        MERGE (m1:Movie {title: "长津湖", released: 2021})
        MERGE (m2:Movie {title: "满江红", released: 2023})
        MERGE (p1:Person {name: "吴京", born: 1974})
        MERGE (p2:Person {name: "易烊千玺", born: 2000})
        MERGE (g1:Genre {name: "动作"})
        MERGE (g2:Genre {name: "历史"})
        MERGE (p1)-[:ACTED_IN {roles: ["伍千里"]}]->(m1)
        MERGE (p2)-[:ACTED_IN {roles: ["伍万里"]}]->(m1)
        MERGE (p2)-[:ACTED_IN {roles: ["孙均"]}]->(m2)
        MERGE (m1)-[:BELONGS_TO]->(g1)
        MERGE (m1)-[:BELONGS_TO]->(g2)
        MERGE (m2)-[:BELONGS_TO]->(g1)
    """)

def find_actors_in_movie(tx, movie_title):
    """查找电影中的演员"""
    result = tx.run("""
        MATCH (m:Movie {title: $movie_title})<-[:ACTED_IN]-(a:Person)
        RETURN a.name AS actor, a.born AS born
    """, movie_title=movie_title)
    return [{"actor": record["actor"], "born": record["born"]} for record in result]

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

with driver.session() as session:
    # 创建知识图谱
    session.execute_write(create_movie_knowledge_graph)
    print("知识图谱创建成功！")

    # 查找电影中的演员
    actors = session.execute_read(find_actors_in_movie, "长津湖")
    print(f"长津湖的演员：{actors}")

    # 查找特定类型的电影
    movies = session.execute_read(find_movies_by_genre, "动作")
    print(f"动作类型的电影：{movies}")

driver.close()

5. 最佳实践 #

5.1 使用参数化查询 #

推荐：使用参数化查询

# 使用参数，安全且高效
tx.run("MATCH (p:Person {name: $name}) RETURN p", name=person_name)

❌ 不推荐：字符串拼接

# 不安全，容易受到注入攻击
tx.run(f"MATCH (p:Person {{name: '{person_name}'}}) RETURN p")

5.2 使用事务 #

推荐：使用事务管理

with driver.session() as session:
    session.execute_write(create_person, "张三", 30)

❌ 不推荐：直接执行

# 不推荐直接执行，应该使用事务
driver.session().run("CREATE (p:Person {name: '张三'})")

5.3 关闭连接 #

推荐：总是关闭连接

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

5.4 错误处理 #

推荐：处理异常

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

6. 小结 #

6.1 核心概念 #

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

6.2 最佳实践 #

1. 使用参数化查询：安全且高效
2. 使用事务管理：确保数据一致性
3. 总是关闭连接：释放资源
4. 处理异常：提高程序健壮性

6.3 常见用法 #

• 使用GraphDatabase.driver()创建连接
• 使用session.execute_write()执行写入操作
• 使用session.execute_read()执行读取操作
• 使用MERGE避免重复数据
• 使用参数化查询提高安全性