Pyneo写入Neo4j关系边数量缺失仅部分插入成功问题咨询

问题背景 #

在大学项目中结合Neo4j、Flask与pyneo开发排班算法时，出现排班关系写入丢失问题：总计330条待插入关系最终仅成功写入91条。

• 插入前后打印待插入列表，确认所有待插入关系均已生成在列表中
• 多次调整事务提交位置验证影响，问题未解决

采用的图数据结构：(w:Worker)-[r:works_during]->(s:Shift)，为关系r设置r.day、r.month、r.year三个属性，同一Worker与Shift之间允许存在多条关联，后续可通过关系属性做筛选。

原实现代码如下：

header = df.columns.tolist()
header.remove("index")
header.remove("worker")
tuplelist = []
for index, row in df.iterrows():
    for i in header:
        worker = self.driver.nodes.match("Worker", id=int(row["worker"])).first()
        if row[i] == 1:
            # Shifts are in the format {day}_{shift_of_day}
            shift_id = str(i).split("_")[1]
            shift_day = str(i).split("_")[0]
            shift = self.driver.nodes.match("Shift", id=int(shift_id)).first()
            rel = Relationship(worker, "works_during", shift)
            rel["day"] = int(shift_day)
            rel["month"] = int(month)
            rel["year"] = int(year)
            tuplelist.append(rel)
print(len(tuplelist))

for i in tuplelist:
    connection = self.driver.begin()
    connection.create(i)
    connection.commit()

问题根因 #

pyneo不存在未公开的特殊机制导致关系丢失，问题出在代码逻辑缺陷：

• 节点匹配无空值校验：循环内调用first()获取Worker、Shift节点时，若匹配条件不命中（如id类型转换错误、对应节点不存在、shift id解析逻辑错误），方法会返回None。此时传入None作为关系的起点/终点创建Relationship对象，pyneo不会抛出显性异常，会在执行create时静默丢弃这条无效关系，不会写入数据库。330条仅成功写入91条，基本可以确定剩余239条关系对应的起止节点存在匹配为空的情况。
• 事务逻辑不合理且无异常捕获：每条关系单独开启、提交一次事务，不仅写入性能极差，且全程没有异常捕获逻辑，写入过程中触发的锁冲突、网络闪断、事务超时等报错会被直接吞掉，无法感知失败原因。

注：pyneo的create()方法默认不做去重判断，该特性只会导致重复写入，不会造成数据丢失，和当前问题无关。

修复方案 #

1. 避免在双层循环内反复查询节点，提前一次性批量拉取所有需要的Worker、Shift节点，存入以id为key的字典，既提升查询性能，也方便做存在性校验
2. 增加节点存在性判断，匹配不到节点时直接打印对应id和关联信息，快速定位缺失数据
3. 把所有关系放到同一个事务内批量提交，增加try/except逻辑，写入失败时回滚事务并打印错误信息

修复后参考代码：

header = df.columns.tolist()
header.remove("index")
header.remove("worker")
tuplelist = []

# 提前批量拉取所有需要的节点，避免循环内重复查库
worker_ids = df["worker"].astype(int).unique().tolist()
worker_map = {
    w["id"]: w
    for w in self.driver.nodes.match("Worker").where("_.id in $ids", ids=worker_ids)
}
shift_ids = list({int(str(i).split("_")[1]) for i in header})
shift_map = {
    s["id"]: s
    for s in self.driver.nodes.match("Shift").where("_.id in $ids", ids=shift_ids)
}

for index, row in df.iterrows():
    worker_id = int(row["worker"])
    if worker_id not in worker_map:
        print(f"写入跳过：缺失Worker节点，id={worker_id}")
        continue
    worker = worker_map[worker_id]
    for i in header:
        if row[i] == 1:
            shift_day = str(i).split("_")[0]
            shift_id = int(str(i).split("_")[1])
            if shift_id not in shift_map:
                print(f"写入跳过：缺失Shift节点，id={shift_id}，对应列：{i}")
                continue
            shift = shift_map[shift_id]
            rel = Relationship(worker, "works_during", shift)
            rel["day"] = int(shift_day)
            rel["month"] = int(month)
            rel["year"] = int(year)
            tuplelist.append(rel)

print(f"待插入关系总数：{len(tuplelist)}")

# 批量提交事务，增加异常捕获和回滚逻辑
connection = self.driver.begin()
try:
    for rel in tuplelist:
        connection.create(rel)
    connection.commit()
    print("关系写入完成")
except Exception as e:
    connection.rollback()
    print(f"写入失败，事务已回滚，错误：{str(e)}")
    raise

内容的提问来源于stack exchange，提问作者Molarix