如何向MongoDB嵌套数组条件执行$push 避免连续相同值并保证原子性

结论 #

上述需求完全可以实现，仅需一次原子性的数据库操作即可覆盖所有要求，无需分两次调用。你需要使用MongoDB 4.2及以上版本支持的更新聚合管道能力，替代原有简单的$push更新逻辑。

核心实现逻辑 #

我们通过聚合管道分阶段完成所有校验和修改：

1. 探针不存在自动创建：先判断probes数组中是否有匹配probe_id的子文档，没有就直接插入新的探针结构
2. 临时插入新读数：将待插入的读数追加到目标探针的readings数组中
3. 排序：对readings数组按when字段升序排序
4. 连续值去重校验：遍历排序后的数组，对比当前项和上一个保留项的value：
   • 两者value不同：直接保留当前项
   • 两者value相同：如果当前项的when更早，替换上一个保留项，否则丢弃当前项
5. 最终用处理后的数组覆盖原readings数组

完整PyMongo代码 #

from bson import ObjectId
from pymongo import MongoClient

client = MongoClient('你的MongoDB连接串')
db = client['你的库名']
collection = db['你的集合名']

# 示例参数，替换为你的实际参数
location_id = ObjectId('6118d887066a0b17c9a4a531')
probe_id = ObjectId('6118d887066a0b17c9a4a530')
new_reading = {"value": 37, "when": "2021-08-12T07:34:30.405Z"}

result = collection.update_one(
    {"_id": location_id},
    [
        # 阶段1：探针不存在则自动创建
        {
            "$set": {
                "probes": {
                    "$cond": {
                        "if": {"$in": [probe_id, "$probes.probe_id"]},
                        "then": "$probes",
                        "else": {
                            "$concatArrays": [
                                "$probes",
                                [{"probe_id": probe_id, "readings": []}]
                            ]
                        }
                    }
                }
            }
        },
        # 阶段2：定位目标探针，处理读数插入、排序、去重逻辑
        {
            "$set": {
                "probes": {
                    "$map": {
                        "input": "$probes",
                        "as": "probe",
                        "in": {
                            "$cond": {
                                "if": {"$eq": ["$$probe.probe_id", probe_id]},
                                "then": {
                                    "probe_id": probe_id,
                                    "readings": {
                                        "$reduce": {
                                            # 插入新读数后按when升序排序（$sortArray要求MongoDB 5.2+）
                                            "input": {
                                                "$sortArray": {
                                                    "input": {"$concatArrays": ["$$probe.readings", [new_reading]]},
                                                    "sortBy": {"when": 1}
                                                }
                                            },
                                            "initialValue": [],
                                            "in": {
                                                "$let": {
                                                    "vars": {
                                                        "last": {"$last": "$$value"},
                                                        "current": "$$this"
                                                    },
                                                    "in": {
                                                        "$cond": {
                                                            # 数组为空或value不重复，直接追加
                                                            "if": {"$or": [
                                                                {"$not": "$$last"},
                                                                {"$ne": ["$$last.value", "$$current.value"]}
                                                            ]},
                                                            "then": {"$concatArrays": ["$$value", ["$$current"]]},
                                                            # value重复，仅当新项时间更早时替换旧项
                                                            "else": {
                                                                "$cond": {
                                                                    "if": {"$lt": ["$$current.when", "$$last.when"]},
                                                                    "then": {"$concatArrays": [{"$slice": ["$$value", 0, -1]}, ["$$current"]]},
                                                                    "else": "$$value"
                                                                }
                                                            }
                                                        }
                                                    }
                                                }
                                            }
                                        }
                                    }
                                },
                                "else": "$$probe"
                            }
                        }
                    }
                }
            }
        }
    ]
)

特性说明 #

• 全程原子性：单条update_one操作，不存在并发冲突问题
• 自动处理探针不存在的场景，无需额外二次调用
• 完全匹配所有需求规则，覆盖你列出的8种插入场景
• 天然保证readings数组始终按when升序排列

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