Python实现图数据库操作_Neo4j核心CRUD与图算法解析
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 都支持流式返回或写回图中。
- 投影图：
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)
