MongoDB实现：不存在则插入，仅当sync_date更晚时更新

我正在构建一个同步系统，数据格式如下：

records = [
    {'_id': 0, 'name': 'John', 'sync_date': '2022-01-01 00:00:00'}, 
    {'_id': 1, 'name': 'Paul', 'sync_date': '2021-11-11 11:11:11'}, 
    {'_id': 2, 'name': 'Anna', 'sync_date': '2012-12-12 12:12:12'}
]

需求明确：

• 若集合中不存在对应_id的记录，直接插入该记录；
• 若存在对应_id的记录，仅当待同步记录的sync_date晚于集合中已有记录的sync_date时，才执行更新操作。

举个例子：如果集合中已有数据：

[
    {'_id': 0, 'name': 'John', 'sync_date': '2000-01-01 00:00:00'}, 
    {'_id': 1, 'name': 'Paul', 'sync_date': '2023-01-01 23:23:23'}
]

那么最终应该：

• 更新John的记录（因为集合里的sync_date更早）；
• 忽略Paul的记录（集合里的sync_date更晚）；
• 插入Anna的记录（集合里无对应_id）。

我尝试用replace_one或update_one加upsert=True时，出现了问题：当集合中已有记录的sync_date更晚时，会误插入新记录。错误代码如下：

for record in records:
    collection.replace_one({
            '_id': record['_id'], 
            'sync_date': {'$lt': record['sync_date']}
            },
        {'upsert': True})

目前的临时方案是分两次查询（先判断是否存在，再决定插入或更新），但效率不高：

for record in records:
    if not collection.find_one({'_id': record['_id']}):
        collection.insert_one(record)
    else:
        collection.update_one(
            {'_id': record['_id'], 'sync_date': {'$lt': record['sync_date']}},
            record)

请问如何构建更高效的查询语句？

1. 修正replace_one写法（单条记录处理） #

你之前的代码错误是参数传递有误：replace_one的第二个参数应该是要替换的文档，upsert=True是作为关键字参数传入的，不是第二个参数。修正后的写法完全符合需求：

for record in records:
    collection.replace_one(
        # 查询条件：匹配_id，且已有记录的sync_date早于待同步记录
        {'_id': record['_id'], 'sync_date': {'$lt': record['sync_date']}},
        # 要替换/插入的文档
        record,
        # 无匹配时自动插入
        upsert=True
    )

逻辑效果：

• 无对应_id：查询条件不匹配，触发upsert=True插入新记录；
• 有对应_id但sync_date更晚：查询条件不匹配，不执行任何操作；
• 有对应_id且sync_date更早：查询条件匹配，执行替换操作。

每个记录仅需一次数据库请求，比临时方案的两次查询高效得多。

2. 批量处理：用bulk_write提升效率 #

如果待同步记录数量较多，推荐使用bulk_write批量执行所有操作，减少网络往返次数，大幅提升同步效率：

from pymongo import UpdateOne

operations = []
for record in records:
    operations.append(
        UpdateOne(
            {'_id': record['_id'], 'sync_date': {'$lt': record['sync_date']}},
            {'$set': record},
            upsert=True
        )
    )

# 一次性执行所有操作
collection.bulk_write(operations)

这里用UpdateOne配合$set更新，效果和replace_one类似，但$set只会更新指定字段（如果后续记录新增其他字段，$set比replace_one更安全，不会覆盖未指定的字段）。

方案优势 #

• 避免临时方案中find_one+insert_one/update_one的两次数据库交互，每个记录仅需一次操作；
• 批量操作bulk_write把多个请求合并成一次，进一步减少网络开销，适合大规模同步场景；
• 逻辑完全由数据库端处理，无需在应用层做额外判断，降低代码复杂度。

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