从 Python 程序发出图数据库指令

Neo4j Python
编程操作实践

这节课不追求一次写很多代码，而是把连接、增查改删、事务和错误处理这条主线走稳。

实践路线：先懂方法，再动手执行

连接

用 Python Driver 建立入口，检查数据库能不能连通。

操作

围绕增、查、改、删练习 `execute_query()` 与参数传递。

事务

把多句写操作放进同一个可靠边界，理解成功与失败要一起处理。

收尾

学习结果处理、异常处理和一个可复用的小脚本结构。

本节数据小图：学生、课程、技能

学生 Student

保存学号、姓名、年级。课堂中用它代表学习者。

课程 Course

保存课程编号、课程名、学时。课堂中用它代表学习资源。

技能 Skill

保存技能名称和类别。课堂中用它连接课程与岗位能力。

关系：学生选择课程，课程培养技能。所有课堂数据都带 lesson_tag，便于安全清理。

环境准备：先把变量写清楚

URI：Neo4j 的连接地址，常见本地地址是 `neo4j://localhost:7687`。
AUTH：认证元组，格式是 `(用户名, 密码)`，课堂示例用占位密码。
DATABASE：目标数据库，建议每次查询都显式写上。
LESSON_TAG：课堂数据标记，后面清理数据时非常有用。

config.py

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your-password")
DATABASE = "neo4j"
LESSON_TAG = "python_practice_01"

连接命令讲解：Driver 是程序总入口

GraphDatabase.driver()

创建 Driver 对象。Driver 负责维护连接信息和连接池，通常一个应用创建一个就够。

verify_connectivity()

只检查是否能和数据库通信，不执行业务查询。适合课前检查和启动自检。

with 语句

用上下文管理器自动关闭 Driver，避免连接资源一直占着不放。

记忆方式：Driver 像“班级总入口”，Session 像“一次办事窗口”，Query 才是具体要办的事。

动手：写一个最小连接检查

connect_check.py

from neo4j import GraphDatabase

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your-password")

with GraphDatabase.driver(URI, auth=AUTH) as driver:
driver.verify_connectivity()
print("Neo4j 连接成功，可以开始实践")

课堂操作重点：只改密码，不改代码结构。先让全班都看到“连接成功”。

小练习：换一种写法保存连接配置

题目

把 `URI`、`AUTH`、`DATABASE` 放入一个 `dict`，再用字典里的值创建 Driver。

要求

仍然调用 `verify_connectivity()`，成功后输出数据库名和“准备完成”。

提示

字典取值写法是 `config["uri"]`，认证信息仍然是二元组。

exercise hint

config = {
    "uri": "neo4j://localhost:7687",
    "auth": ("neo4j", "your-password"),
    "database": "neo4j"
}

查询入口讲解：execute_query() 适合多数课堂操作

query

第一项通常是 Cypher 字符串，描述要对图数据库做什么。

parameters_

把变量值放进字典，Cypher 中用 `$变量名` 接收，避免字符串拼接。

database_

指定目标数据库。大量查询时显式指定数据库更清晰，也减少额外判断。

常见返回：records 是结果行，summary 是执行摘要，keys 是字段名。

动手：用 Python 安全清理课堂数据

reset_lesson_data.py

from neo4j import GraphDatabase

def reset_lesson_data(driver, tag):
    query = """
    MATCH (n {lesson_tag: $tag})
    DETACH DELETE n
    """
    summary = driver.execute_query(
        query,
        parameters_={"tag": tag},
        database_=DATABASE,
    ).summary
    print("删除节点数：", summary.counters.nodes_deleted)

课堂操作重点：只删除带 lesson_tag 的节点，避免误删已有数据。

小练习：为岗位数据写安全清理函数

题目

把清理对象换成岗位实践数据，标记改为 `job_practice_01`。

结构

函数名改成 `reset_job_data(driver, tag)`，内部仍然使用 `MATCH ... DETACH DELETE`。

提醒

不要写 `MATCH (n) DETACH DELETE n`，课堂环境里也要养成安全习惯。

新增命令讲解：CREATE、MERGE、SET

CREATE

直接创建新节点或关系。适合确定不会重复的数据，但课堂中较少单独使用。

MERGE

先匹配，找不到再创建。适合学生、课程、技能这类有唯一标识的数据。

SET

设置或更新属性。常和 `MERGE` 配合使用，保证重复执行也能得到一致结果。

课堂新增数据尽量用 MERGE + SET，这样重复运行代码时不容易制造一堆重复节点。

动手：新增学生、课程、技能节点

create_nodes.py

def upsert_base_nodes(driver):
    query = """
    MERGE (s:Student {sid: $sid})
    SET s.name = $student_name,
        s.grade = $grade,
        s.lesson_tag = $tag
    MERGE (c:Course {cid: $cid})
    SET c.name = $course_name,
        c.hours = $hours,
        c.lesson_tag = $tag
    MERGE (k:Skill {name: $skill_name})
    SET k.category = $category,
        k.lesson_tag = $tag
    """
    params = {
        "sid": "S202401", "student_name": "李明", "grade": "大二",
        "cid": "C101", "course_name": "图数据库基础", "hours": 48,
        "skill_name": "Cypher 查询", "category": "数据查询",
        "tag": LESSON_TAG
    }
    driver.execute_query(query, parameters_=params, database_=DATABASE)

动手：新增关系，让小图连起来

create_relationships.py

def connect_base_graph(driver):
    query = """
    MATCH (s:Student {sid: $sid})
    MATCH (c:Course {cid: $cid})
    MATCH (k:Skill {name: $skill_name})
    MERGE (s)-[r1:SELECTED]->(c)
    SET r1.lesson_tag = $tag, r1.source = "课堂实践"
    MERGE (c)-[r2:TEACHES]->(k)
    SET r2.lesson_tag = $tag, r2.level = "入门"
    """
    driver.execute_query(
        query,
        parameters_={"sid": "S202401", "cid": "C101", "skill_name": "Cypher 查询", "tag": LESSON_TAG},
        database_=DATABASE,
    )

关系也可以有属性。比如来源、等级、时间，这些都能帮助后续分析。

小练习：换成岗位、企业、技能小图

节点变化

把 `Student`、`Course`、`Skill` 换成 `Company`、`Job`、`Skill`。

关系变化

企业 `PUBLISHES` 岗位，岗位 `REQUIRES` 技能。

数据变化

企业名用“星河数据服务”，岗位用“数据标注工程师”，技能用“数据清洗”。

exercise shape

MERGE (co:Company {name: $company})
MERGE (j:Job {title: $job})
MERGE (sk:Skill {name: $skill})
MERGE (co)-[:PUBLISHES]->(j)
MERGE (j)-[:REQUIRES]->(sk)

读取命令讲解：MATCH、WHERE、RETURN

MATCH

描述要匹配的图模式。可以查节点，也可以沿着关系查路径。

WHERE

补充过滤条件，例如只查某个学生、某类技能、某个课堂标记。

RETURN

指定返回字段。只返回课堂需要展示的字段，不要把完整节点全搬回来。

读取常配套：ORDER BY 排序，LIMIT 控制返回数量。

动手：查询学生选择了哪些课程

read_courses.py

def find_student_courses(driver, sid):
    query = """
    MATCH (s:Student {sid: $sid})-[:SELECTED]->(c:Course)
    WHERE s.lesson_tag = $tag
    RETURN s.name AS student, c.name AS course, c.hours AS hours
    ORDER BY c.hours DESC
    LIMIT 5
    """
    records, summary, keys = driver.execute_query(
        query,
        parameters_={"sid": sid, "tag": LESSON_TAG},
        database_=DATABASE,
    )
    for record in records:
        print(record["student"], "选择了", record["course"], record["hours"], "学时")

动手：沿着路径查“课程培养什么技能”

read_path.py

def find_course_skills(driver, cid):
    query = """
    MATCH (c:Course {cid: $cid})-[r:TEACHES]->(k:Skill)
    WHERE c.lesson_tag = $tag
    RETURN c.name AS course,
           k.name AS skill,
           r.level AS level
    ORDER BY skill
    """
    records, _, _ = driver.execute_query(
        query,
        parameters_={"cid": cid, "tag": LESSON_TAG},
        database_=DATABASE,
    )
    return [record.data() for record in records]

record.data() 会把一行结果转成 Python 字典，后面做表格或接口返回更方便。

小练习：查询岗位需要哪些技能

题目

查询“数据标注工程师”岗位需要的技能，返回岗位名、技能名、技能类别。

结构

沿用 `MATCH (j:Job)-[:REQUIRES]->(sk:Skill)` 的路径结构。

限制

必须带 `lesson_tag` 过滤，并使用 `LIMIT 10` 控制结果数量。

修改命令讲解：SET 与 REMOVE

SET 属性

给节点或关系新增、覆盖属性。比如修改课程学时、技能等级、岗位状态。

SET 标签

可以给节点补充标签，例如把重点课程标记为 `KeyCourse`。

REMOVE

移除某个属性或标签。课堂中先少用，重点理解修改属性。

修改前最好能先 MATCH 精准定位，再 SET 目标字段。

动手：修改课程学时和技能等级

update_course.py

def update_course(driver, cid, hours, level):
    query = """
    MATCH (c:Course {cid: $cid})-[r:TEACHES]->(:Skill)
    WHERE c.lesson_tag = $tag
    SET c.hours = $hours,
        c.updated_by = "python",
        r.level = $level
    RETURN c.name AS course, c.hours AS hours, r.level AS level
    """
    records, summary, _ = driver.execute_query(
        query,
        parameters_={"cid": cid, "hours": hours, "level": level, "tag": LESSON_TAG},
        database_=DATABASE,
    )
    print("修改属性数：", summary.counters.properties_set)
    print(records[0].data())

小练习：修改岗位状态

题目

把“数据标注工程师”岗位的 `status` 设置为 `招聘中`，把 `salary_range` 设置为 `4K-6K`。

方法

使用 `MATCH (j:Job {title: $title})` 定位，再用 `SET` 修改属性。

输出

返回岗位名、状态、薪资范围，并打印 `summary.counters.properties_set`。

删除命令讲解：先想清楚删什么

DELETE 关系

如果只想取消连接，先匹配关系变量，再删除关系。

DELETE 节点

节点没有关系时才能直接删除。适合清理孤立测试节点。

DETACH DELETE

连同节点上的关系一起删除。课堂中只对带标记的数据使用。

删除是数据库操作里最需要谨慎的一类。课堂里始终用参数和 lesson_tag 加保护。

动手：删除一条测试关系

delete_relation.py

def delete_selected_relation(driver, sid, cid):
    query = """
    MATCH (s:Student {sid: $sid})-[r:SELECTED]->(c:Course {cid: $cid})
    WHERE r.lesson_tag = $tag
    DELETE r
    """
    summary = driver.execute_query(
        query,
        parameters_={"sid": sid, "cid": cid, "tag": LESSON_TAG},
        database_=DATABASE,
    ).summary
    print("删除关系数：", summary.counters.relationships_deleted)

先删关系，再决定要不要删节点。这样更容易控制影响范围。

小练习：删除岗位和技能之间的测试关系

题目

删除“数据标注工程师”到“数据清洗”的 `REQUIRES` 关系。

保护条件

必须要求关系上有 `lesson_tag = job_practice_01`。

输出

打印 `summary.counters.relationships_deleted`，确认只删除一条或少量测试关系。

事务命令讲解：execute_write() 把多步放进边界

session()

创建一次数据库工作窗口，推荐写上 `database=DATABASE`。

execute_write()

接收一个事务函数。驱动会在合适情况下处理重试和提交。

tx.run()

事务函数内部用 `tx.run()` 执行 Cypher，并在函数里处理结果。

事务适合“几件事必须一起成功”的场景，例如创建选课关系并记录选课时间。

动手：用事务完成选课

transaction_enroll.py

def enroll_course(tx, sid, cid, tag):
    query = """
    MATCH (s:Student {sid: $sid})
    MATCH (c:Course {cid: $cid})
    MERGE (s)-[r:SELECTED]->(c)
    SET r.lesson_tag = $tag,
        r.status = "已选",
        r.created_by = "transaction"
    RETURN s.name AS student, c.name AS course
    """
    result = tx.run(query, sid=sid, cid=cid, tag=tag)
    return result.single().data()

with driver.session(database=DATABASE) as session:
    data = session.execute_write(enroll_course, "S202401", "C101", LESSON_TAG)
    print(data)

小练习：用事务维护岗位技能关系

题目

写 `attach_required_skill(tx, job_title, skill_name, tag)`，把岗位和技能连起来。

事务要求

用 `session.execute_write()` 调用事务函数，不直接在外面拼接 Cypher。

返回结果

返回岗位名和技能名，确认关系已经创建或复用。

结果处理讲解：records、summary、keys

records

查询结果行列表。每个 `record` 可以用字段名取值，也可以转成字典。

summary

执行摘要，包含计数器和耗时。写操作后经常看 counters。

keys

返回字段名列表。适合调试，也适合做简单表格输出。

面向应用开发时，不要只看控制台，要把结果整理成 Python 能继续处理的数据结构。

动手：把查询结果打印成小表格

format_records.py

def print_records(title, records):
    print("\n" + title)
    print("-" * 40)
    for record in records:
        row = record.data()
        print(" | ".join(f"{k}: {v}" for k, v in row.items()))

records, summary, keys = driver.execute_query(
    """
    MATCH (:Course)-[:TEACHES]->(k:Skill)
    WHERE k.lesson_tag = $tag
    RETURN k.name AS skill, k.category AS category
    ORDER BY skill
    """,
    parameters_={"tag": LESSON_TAG},
    database_=DATABASE,
)
print_records("课程培养的技能", records)

异常处理讲解：程序要把问题说清楚

连接类问题

数据库没启动、端口不通、URI 写错，通常在连接或首次查询时暴露。

认证类问题

用户名、密码、权限错误。提示学生先检查密码，再检查数据库名。

语句类问题

标签、属性、参数名拼错。错误信息里常能看到出问题的位置。

实践课里不追求把所有异常都处理完，重点是不要让程序“静静失败”。

动手：写一个安全执行函数

safe_execute.py

from neo4j.exceptions import Neo4jError, ServiceUnavailable, AuthError

def safe_execute(driver, query, params):
    try:
        records, summary, keys = driver.execute_query(
            query,
            parameters_=params,
            database_=DATABASE,
        )
        print("执行成功，返回字段：", keys)
        return [record.data() for record in records]
    except AuthError:
        print("认证失败：请检查用户名或密码")
    except ServiceUnavailable:
        print("连接失败：请检查 Neo4j 是否启动")
    except Neo4jError as error:
        print("Neo4j 执行错误：", error)
    return []

性能与安全提醒：课堂代码也要有好习惯

指定数据库

每次查询写 `database_=DATABASE` 或 `session(database=DATABASE)`。

限制结果

查询展示数据时加 `LIMIT`，不要一次把整张图搬到 Python。

使用参数

外部输入放进 `parameters_`，不要把用户输入直接拼到 Cypher 字符串里。

这三条不是高级技巧，是从第一节实践课就应该养成的基本习惯。

把代码收成一个小脚本

main.py

from neo4j import GraphDatabase

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "your-password")
DATABASE = "neo4j"
LESSON_TAG = "python_practice_01"

def main():
    with GraphDatabase.driver(URI, auth=AUTH) as driver:
        driver.verify_connectivity()
        reset_lesson_data(driver, LESSON_TAG)
        upsert_base_nodes(driver)
        connect_base_graph(driver)
        find_student_courses(driver, "S202401")

if __name__ == "__main__":
    main()

课堂上可以分段运行，课后建议收成 main()，方便复现和检查。

综合练习：换数据，但不换结构

新数据

企业：云帆科技；岗位：初级数据分析助理；技能：Python 清洗、可视化表达。

必做操作

连接检查、安全清理、新增节点、新增关系、查询岗位技能、修改岗位状态。

加分操作

用事务一次性维护岗位和技能关系，并打印 `summary` 或返回字典结果。

提交形式：一个 Python 文件，控制台能看到每一步的关键输出。

综合练习检查点

连接：能看到连接成功提示，不把密码写进截图或提交说明。
新增：重复运行脚本不会出现大量重复节点。
查询：至少返回岗位、技能、关系属性中的两个字段。
修改：能打印修改后的岗位状态或技能等级。
删除：只清理带课堂标记的数据，不影响其他同学的数据。
代码：函数命名清楚，参数通过 `parameters_` 传入。

常见问题快速排查

连不上

先确认 Neo4j 服务已启动，再检查 URI、端口 7687、用户名和密码。

查不到

检查标签名、属性名、参数名是否一致，尤其是 `$sid` 与 `parameters_` 里的键。

重复了

新增实体时优先用 `MERGE`，并选择稳定的唯一字段，例如学号、课程编号、岗位标题。

实践收束：今天真正练会什么

会连接

能用 Python Driver 创建入口，并检查 Neo4j 是否可用。

会操作

能用 Python 调用 Cypher 完成增、查、改、删的核心流程。

会保护

知道用参数、课堂标记、事务和异常处理降低操作风险。

下次再进入更复杂的批量导入、扩展过程或运维命令时，今天这套 Python 调用方式仍然是基础。

Neo4j Python编程操作实践