很多人在接 Neo4j 的时候第一反应是打开官方 REST 接口POST /db/data/transaction/commit跑一遍 Cypher把返回的大 JSON 直接透传给前端。我也这么干过直到数据库从 3.x 升到 5.xREST API 被官方逐步边缘化Bolt 成了所有语言驱动的默认协议才反应过来连接方式换了之后查询结果的对象模型完全不同了。这篇文章就围绕一个真实需求展开neo4j 数据库使用 Bolt 协议拿到查询结果后如何转换成适合 REST API 返回的 JSON 格式。我会从 Bolt 返回对象的结构讲起给出可复用的 Python 序列化函数、FastAPI 接入示例再聊一些我在实际项目里踩过的坑。如果你正在把一个旧 Neo4j HTTP 接口逐步迁移到官方 Driver或者单纯被TypeError: Object of type Node is not JSON serializable折磨过看完你应该能直接“抄作业”。1. 为什么 REST 接口和 Bolt 拿到的结果长得不一样1.1 旧版 REST API 的返回结构是套娃式的以前用curl调 Neo4j HTTP API 时返回结构固定是“外层结果 列名 数据行”的形式。比如执行一条最简单的MATCH (p:Person) RETURN p{ results: [ { columns: [p], data: [ { row: [ { name: 张三, age: 30 } ], meta: [ { id: 123, type: node, deleted: false } ] } ] } ], errors: [] }这种格式本身没问题但如果你想封装一个自己的业务 API比如返回{user: {...}}就必须先剥掉三层壳results、data、row而且每层结构还不一样。更别扭的是节点的属性值和元数据是分开放的row里明明是属性 mapmeta里才放节点 ID前端拿到的状态和数据库状态总是隔着一层。1.2 Bolt 驱动返回的是“对象”不是“字典”换成 Bolt 之后官方 Driver 不再走 HTTP而是通过原生二进制协议和数据库通信。拿 Python 驱动举例from neo4j import GraphDatabase driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, password)) query MATCH (p:Person) RETURN p LIMIT 1 with driver.session() as session: record session.run(query).single() p record[p] print(p)打印出来的东西长这样Node element_id4:abcdef... labels{Person} properties{name: 张三, age: 30}record[p]返回的是neo4j.graph.Node对象不是dict。Node 虽然实现了类似 Mapping 的接口可以p[name]取属性但它还有.labels、.element_id这类额外的图结构信息。你直接扔给json.dumps就会报TypeError: Object of type Node is not JSON serializable所以这里的核心矛盾就出来了Bolt 的查询结果原生是对象模型而 REST API 的输出要求是纯 JSON。所谓的“转换 RestApi 请求结果格式”本质上是在 Bolt 驱动层加一个序列化适配层把节点、关系、路径拆成普通 Python 字典再统一转 JSON。2. 先弄清楚 Bolt 查询结果的真实形状再谈转换2.1 Record、Node、Relationship、Path 四类对象的核心结构在写转换函数之前得先知道我们面对的四类对象各自长什么样。RecordBolt 中一行查询结果的封装。它有两个关键操作record.keys()返回列名列表。record.data()返回{字段名: 值}字典。但注意record.data()不会递归展开节点节点依然是 Node 对象。Node代表图里的一个节点。常见属性有.labelsfrozenset节点的标签集合。.element_id新版本驱动里的唯一 ID形式类似4:xxxxxx。.properties属性字典也可以直接dict(node)获取。传统的.id在部分版本已经废弃尽量别用。Relationship代表一条边。它同样有.type关系的类型、.start_node、.end_node、.element_id同时也可以dict(rel)拿属性。Path代表一条路径。它内部是节点和关系的交替序列比较特殊的一点是path.nodes会扁平地返回路径上所有节点但这样会丢失关系方向。理解了这四种类型转换逻辑就很清晰了递归遍历记录中的每个值遇到 Node、Relationship、Path 时就拆成 dict遇到普通类型就原样保留。2.2 一个记录里既有节点又有聚合值怎么设计通用转换规则实际查询很少只返回一个纯节点更多是这种混合结果MATCH (p:Person)-[r:ACTED_IN]-(m:Movie) WHERE p.name $name RETURN p, collect(m) AS movies这行结果里p是 Nodemovies是一个 Node 列表。如果你只做一层dict(record[p])那movies里仍然是节点对象序列化照样炸。所以我建议做一套通用递归转换函数而不是针对每个查询写专门逻辑。判断逻辑只有一条值是 Node / Relationship / Path走对应拆分逻辑。值是 list / dict递归处理内部元素。值是时间类型、字节类型转成统一字符串或列表。其余普通类型int、str、bool、None直接返回。2.3 从节点里取哪些字段建议保留 element_id 而不是数字 id这里有个很容易踩坑的设计决策转换后要不要把节点 ID 放进去早期 Neo4j 的 HTTP 返回里带的是数字 ID也就是meta.id。但 Neo4j 4.0 之后内部 ID 的稳定性和可重用性存在争议官方主推element_id。Bolt 驱动里拿到的node.element_id是包含了节点类型信息的字符串比如4:abc123。我的经验是如果 API 只给内部服务自己用保留 element_id 作为临时索引字段即可如果 API 是要给前端长期使用的建议用业务自己的主键字段比如 person_code、movie_id作为数据标识不要暴露数据库内部 ID。数据库 ID 不是业务 ID一旦做数据迁移或备份恢复数字 ID 很可能发生变化。3. 用 Python FastAPI neo4j 驱动搭建转换链路的落地代码3.1 环境准备驱动和项目的选择先准备一个本地 Neo4j 实例我用 Docker 跑一个 5.x 版本的docker run --name neo4j-test \ -p 7474:7474 \ -p 7687:7687 \ -e NEO4J_AUTHneo4j/test123456 \ -d neo4j:5.147474 是浏览器控制台端口7687 是 Bolt 协议端口连接字符串用bolt://localhost:7687。Python 环境这边只需要两个库pip install neo4j fastapi uvicornFastAPI 用来起一个测试接口neo4j 就是官方驱动。如果你不想引 FastAPI纯用 Flask 甚至直接用json.dumps看输出也行核心转换逻辑是一样的。3.2 核心转换函数从 Record 到 JSON 的递归序列化我的转换函数分两层。第一层处理单个值第二层处理一整条 recordfrom neo4j.graph import Node, Relationship, Path from datetime import datetime, date, time from typing import Any def convert_value(value: Any) - Any: 递归转换 Bolt 驱动器返回的值使其可被 json.dumps 序列化。 if value is None: return None if isinstance(value, Node): return { element_id: value.element_id, labels: list(value.labels), # 这里直接展开了属性也可以改成 {properties: dict(value)} **dict(value), } if isinstance(value, Relationship): return { element_id: value.element_id, type: value.type, start: value.start_node.element_id, end: value.end_node.element_id, **dict(value), } if isinstance(value, Path): # 路径转换成节点列表 关系列表见 4.2 的详细处理 return path_to_chain(value) if isinstance(value, list): return [convert_value(item) for item in value] if isinstance(value, dict): return {key: convert_value(val) for key, val in value.items()} if isinstance(value, (datetime, date, time)): return value.isoformat() # 其他基础类型直接返回 return value def record_to_dict(record) - dict: 把一行 Cypher 记录转成普通 dict return {key: convert_value(record[key]) for key in record.keys()}这段代码的核心是递归因为 Neo4j 返回的数据经常嵌套。比如一个属性里可能存了数组数组里又可能存了 Map或者用collect()收集了一堆节点。递归处理能让转换逻辑自动覆盖这些情况后续加新查询也不需要改函数。3.3 实现一个“人物信息”查询接口看到完整返回效果接下来把它接到 FastAPI 上。我建一张简单的测试数据CREATE (p:Person {name: 张伟, age: 30})-[r:ACTED_IN {year: 2021}]-(m:Movie {title: 城市边缘, released: 2021})接口逻辑如下from fastapi import FastAPI from neo4j import GraphDatabase app FastAPI() driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, test123456)) app.get(/persons/{name}/details) def person_details(name: str): query MATCH (p:Person)-[r:ACTED_IN]-(m:Movie) WHERE p.name $name RETURN p, collect(m) AS movies with driver.session() as session: record session.run(query, namename).single() if record is None: return {error: person not found} result record_to_dict(record) return result启动服务之后访问/persons/张伟/details输出是这样{ p: { element_id: 4:xxxx, labels: [Person], name: 张伟, age: 30 }, movies: [ { element_id: 4:yyyy, labels: [Movie], title: 城市边缘, released: 2021 } ] }可以看到collect(m)形成的节点数组也被正确展开成了 JSON 数组。前端拿到的就是干净的业务数据不再需要关心 Neo4j 内部的对象类型。3.4 用 json.dumps 的默认 encoder 也能做两种方案对比除了手写递归还有一种做法给json.dumps传一个自定义JSONEncoder。import json from neo4j.graph import Node, Relationship, Path class Neo4jJSONEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, Node): return { element_id: obj.element_id, labels: list(obj.labels), **dict(obj), } if isinstance(obj, Relationship): return { element_id: obj.element_id, type: obj.type, **dict(obj), } if isinstance(obj, Path): return [Neo4jJSONEncoder().default(n) for n in obj.nodes] return super().default(obj)两种方案的区别在于手写递归函数灵活比如路径想拆成什么结构你说了算时间类型想转成什么格式也由你控制。自定义 JSONEncoder侵入小只需要改json.dumps调用处。但如果嵌套在有 FastAPI 这种自带 JSON 序列化的框架里需要额外配置 FastAPI 的默认 encoder反而更麻烦。我实际项目中偏向用递归函数因为在 FastAPI 的响应模型里你返回的 dict 会经过框架自己的jsonable_encoder没法直接复用JSONEncoder。手写转换函数的好处是可以在返回前精确控制数据形态便于调试和打日志。4. 实战中绕不开的细节嵌套列表、路径、时间类型怎么处理4.1 collect() / unwind 之后的列表数据需要按行转数组用collect(m)聚合节点时每个节点依然会被convert_value递归处理成字典。但如果 Cypher 里用了unwind结果会变成多行记录每个 record 单独转换再由 FastAPI 或你的路由把多行记录组装成列表。这一点会在接口层出现app.get(/persons) def all_persons(): query MATCH (p:Person)-[r:ACTED_IN]-(m:Movie) RETURN p.name AS name, m.title AS title with driver.session() as session: result session.run(query) data [record_to_dict(r) for r in result] return {data: data}这里每行record都是{name: ..., title: ...}前端拿到的就是一个扁平数组。聚合和unwind的返回形态差异很大建议在接口层提前定好统一 schema不要直接透传行数不确定的结果。4.2 路径 path 对象别慌拆成节点数组最直观Path 的转换容易让人卡住。因为Path既不是 Node 也不是纯列表单独把它转成字符串毫无意义。我常用的策略是拆成“节点链 关系链”两步def path_to_chain(path: Path) - dict: nodes [] rels [] for item in path: if isinstance(item, Node): nodes.append(convert_value(item)) elif isinstance(item, Relationship): rels.append({ element_id: item.element_id, type: item.type, from: item.start_node.element_id, to: item.end_node.element_id, **dict(item), }) return {nodes: nodes, relationships: rels}这样做的好处是前端既能拿到整条路径上的节点属性也能保留关系方向。如果只需要最短路径上的名字列表前端从nodes数组里取name字段就行不需要自己解析复杂对象。4.3 时间字段和日期字段要给 API 约定好一个统一格式Neo4j 的时间类型比较特殊。它返回的可能是neo4j.time.DateTime也可能是标准库datetime.datetime取决于你是否配置了时区。直接扔给 JSON 序列化会报错标准做法是转成字符串。我在转换函数里统一用.isoformat()if isinstance(value, (datetime, date, time)): return value.isoformat()但这里有个隐藏问题Neo4j 时间类型里带时区信息isoformat()会输出带08:00后缀的字符串。如果你的 API 调用方统一约定用 UTC那就要在读取属性时先自己做时区换算。建议在接口文档里明确写清楚所有日期时间字段统一返回 ISO 8601 字符串。这样既避免了datetime类型无法序列化的问题也减少前端自行解析时间的成本。5. 使用 Bolt 会话的一些性能和稳定性建议5.1 连接池、fetch_size 与事务边界的取舍不要每次请求都新建 Driver。GraphDatabase.driver()内部维护了连接池一般放在应用生命周期里复用driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, test123456))FastAPI 里可以用asynccontextmanager管理生命周期核心同样是把driver搞成单例。另外session.run()在数据量较大时会按批次拉取结果默认fetch_size是 1000 条。对于MATCH ... RETURN这种可能返回上万行的查询建议一次性result.values()前注意内存占用。如果只是需要前几十条结果可以显式设置session.run(query).fetch(100)避免把整个结果集一次性都拉进内存。5.2 查询一定要带参数别用 f-string 拼 Cypher这个我在 code review 里见得太多了。用 f-string 直接拼参数进去不仅容易报废引号还存在注入风险。既然 Bolt 驱动原生支持参数化就老老实实用参数# 正确写法 session.run(MATCH (p:Person) WHERE p.name $name RETURN p, nameuser_input) # 错误写法 session.run(fMATCH (p:Person) WHERE p.name {user_input} RETURN p)参数类型也尽量控制数字就是数字字符串就是字符串。如果参数可能为NoneBolt 驱动也能正确处理但有些 Cypher 版本对WHERE p.name $name遇到None时的行为可能不符合预期可以用IS NULL显式判断。5.3 只读查询用 read_transaction写操作再开写事务一个性能优化点用事务函数时只读查询要声明为read事务写操作用write事务def get_person(tx, name): result tx.run(MATCH (p:Person) WHERE p.name $name RETURN p, namename) return [record_to_dict(r) for r in result] with driver.session() as session: data session.execute_read(get_person, name)为什么不建议直接session.run()一把梭read_transaction和execute_read让驱动知道你会执行只读 Cypher便于它在集群环境里把请求路由到只读副本减少主库压力。即使你用的是单机实例也保持了写操作和读操作边界清晰的好习惯。6. 常见问题倒排速查表6.1 TypeError: Object of type Node is not JSON serializable最常见的问题。原因就是json.dumps()直接接收了 Bolt 返回的节点对象。解决办法就是使用convert_value递归转换或者给JSONEncoder增加 Node 类型分支。不要试图在str(Node)之后做字符串替换那种黑科技不可维护。6.2 返回不了 neo4j.time.DateTime 类型怎么办检查转换函数里是否包含了时间类型的判断。注意除了标准库datetime还要考虑neo4j.time.DateTime、neo4j.time.Date、neo4j.time.Time。最稳妥的办法是在convert_value里针对这几个类都做isoformat()或者统一转成str(value)。建议实际打印一下因为不同驱动版本返回的类型不一样。6.3 节点属性中存在非常规键名要不要处理Neo4j 允许属性键包含空格和特殊字符比如first name。这样的属性在dict(node)里也会原样保留键名。如果前端对 key 有 camelCase 或 snake_case 的规范要求需要在转换层做 key 映射。但这个不建议在通用转换函数里做而是为每个接口单独定义 schema或者用 Pydantic 模型做字段映射。6.4 网页端跑不通 bolt:// 地址怎么办浏览器里的 JS 客户端无法直接通过bolt://协议连接数据库需要在后端封装 REST API。这正是本文讨论的场景。做法有两种后端用 Python / Java / Go Driver 连接 Neo4j自己提供 HTTP JSON 接口。直接在 Neo4j 同机部署一个轻量网关服务。不要试图在前端连接数据库密码会直接暴露在浏览器里等于裸奔。症状可能原因解决方式TypeError: Node is not JSON serializable没做递归转换使用 record_to_dictTransientError: ... fetch结果集过大控制 fetch_size 或分页Too many open connectionsDriver 没复用设置全局单例 driverdatetime序列化失败未处理时间类型convert_value 加.isoformat()旧 REST 接口 404新版禁用 HTTP API改用 Bolt 驱动7. 我的一点实操经验两种转换思路的适用边界7.1 转换层应该放在哪越靠近驱动越好我最初把所有转换逻辑都堆在 FastAPI 路由里结果每个接口都有一段重复的convert_value。后来我把转换逻辑封装成了neo4j_utils.py里的工具函数路由层只返回业务结构。这个改动让后续新增接口的时间明显缩短。建议你把convert_value和record_to_dict单独放模块不要和具体业务混在一起。7.2 通用递归转换好还是按接口定制好这是一个取舍问题。通用递归转换能覆盖 80% 的场景但遇到需要隐藏某些内部属性时就很尴尬。比如某个节点含internal_code你不想让前端看到通用递归函数却会把它原样带出。我的做法是通用转换作为兜底然后在接口层用 Pydantic 模型或手动 dict 做第二层过滤。这样既保留递归转换的便捷又能满足不同接口对字段的差异化需求。7.3 给所有节点加 element_id 是必要的吗如果你的 API 后面还要做 update 操作比如“用户点了按钮要把某个人物的年龄改掉”那确实需要拿一个标识字段回来。建议用element_id作为临时标识写 Cypher 时用WHERE elementId(p) $id定位节点。但如果你只是展示数据把element_id暴露给前端没有实际价值还会让 JSON 变得冗长。很多前端工程师看到4:xxxxx这种字符串会误以为是 ID 规则容易产生困惑。我的习惯是展示型接口去掉element_id管理型接口保留element_id。这样更贴近业务语言的表达。7.4 最后一个小技巧先用record.data()打印原始形状再写转换函数不要上来就写递归。建议先写个临时脚本with driver.session() as session: record session.run(query).single() print(record.data()) print(type(record[p]))看清楚了返回的类型和嵌套层级再决定转换函数该加哪些分支。这个步骤能帮你省掉很多调试时间。尤其是 Cypher 里加了collect、unwind、reduce这类操作之后实际返回的数据形状经常和直觉不一样提前打印是最“笨”但最有效的方式。