Neo4j是主流原生图数据库,Python通过官方驱动高效实现CRUD、调用GDS算法;需复用Driver单例、参数化Cypher、UNWIND批量操作、GDS图投影调用、捕获Neo4jError并用EXPLAIN调试。

Neo4j 是目前最主流的原生图数据库,Python 通过 neo4j 官方驱动(neo4j==5.x)可高效完成节点/关系的增删改查,并直接调用内置图算法库(Graph Data Science Library, GDS)。关键在于:连接要稳、Cypher 要准、事务要控、算法要选对。
连接与会话管理:避免连接泄漏和超时
使用 Driver 单例复用连接池,不每次新建;通过 session() 获取会话,务必显式关闭或用上下文管理器。生产环境建议配置连接超时、最大连接数和加密开关。
- 推荐写法:with driver.session(database="neo4j") as session: 自动释放资源
- 禁用加密(仅开发):
driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"), encrypted=False) - 超时设置:
driver = GraphDatabase.driver(..., connection_timeout=3.0, max_connection_lifetime=3600)
CRUD操作:用参数化Cypher防注入,批量用UNWIND提效
所有写操作必须参数化,禁止字符串拼接;单条数据用 session.run(),批量导入优先用 UNWIND + 参数列表,比循环执行快 10 倍以上。
- 创建节点:
session.run("CREATE (n:Person {name: $name, age: $age})", name="Alice", age=30) - 批量创建:
session.run("UNWIND $data AS row CREATE (p:Product) SET p += row", data=[{"id":1,"price":99}, {"id":2,"price":199}]) - 关联查询(带属性):
session.run("MATCH (a:User)-[r:BOUGHT]->(b:Item) WHERE r.amount > $min RETURN a.name, b.title", min=5) - 删除带约束:
session.run("MATCH (n:Temp) DETACH DELETE n")(DETACH 确保关系一并清理)
图算法调用:GDS插件需预装,算法结果转DataFrame再分析
GDS 不是 Neo4j 内置模块,需单独安装插件并重启服务;算法运行在图投影(graph projection)上,非直接查原始数据。常用算法如 PageRank、Louvain、ShortestPath 都支持流式返回或写回图中。
立即学习“Python免费学习笔记(深入)”;
- 投影图:
session.run("CALL gds.graph.project('myGraph', 'Person', 'FRIEND_OF')") - 运行PageRank:
result = session.run("CALL gds.pageRank.stream('myGraph') YIELD nodeId, score RETURN gds.util.asNode(nodeId).name AS name, score ORDER BY score DESC LIMIT 5") - 结果转 pandas:
import pandas as pd; df = pd.DataFrame([r.data() for r in result]) - 写回节点属性:
session.run("CALL gds.pageRank.write('myGraph', {writeProperty: 'pagerank'})")
错误处理与调试:捕获Neo4jError,开启日志看执行计划
Neo4j 抛出的是 neo4j.exceptions.Neo4jError 及其子类(如 ConstraintError、ClientError),不要用通用 Exception 捕获。调试慢查询时,用 EXPLAIN 或 PROFILE 前缀查看 Cypher 执行计划。
- 捕获唯一约束冲突:
except neo4j.exceptions.ConstraintError as e: print("重复键冲突:", e.message) - 查看执行计划:
session.run("EXPLAIN MATCH (a:User)-[:FRIEND_OF*2]-(b) RETURN count(*)") - 启用驱动日志:
import logging; logging.getLogger("neo4j").setLevel(logging.DEBUG)










