Python操作MongoDB如何简洁实现带时间判断的整记录条件更新

完全可以把时间比较逻辑全部下推到MongoDB侧执行，不需要在Python端先查库再判断，单原子操作就能覆盖你要的三个场景，还能解决原代码两次请求带来的竞态问题。

核心逻辑 #

你要的三个规则可以通过「带聚合管道的update_one + upsert=True」一次性实现，全程只需要和MongoDB交互一次，没有Python侧的冗余分支判断：

• 过滤条件用三个唯一字段stCode、skCode、batchCode匹配已有记录
• 开启upsert=True，匹配不到记录时自动插入新数据
• 用MongoDB聚合表达式在库侧做时间判断：仅当记录不存在、或传入记录的ts大于库内值时，才更新整条记录，否则保留原数据不动

注：你当前库存的ts是YYYY/MM/DD HH:MM:SS格式的定长字符串，字典序和时间顺序完全一致，不需要额外转成时间类型就能直接比较，省掉格式转换的开销。

可直接运行的优化代码 #

首先修正了你原代码里的两个小问题：payload里ts键漏了引号、更新条件里错写的storeCode/skuCode和入参字段不统一的问题。

from datetime import datetime

# 入参payload
payload = {'stCode': 'ABC1', 'skCode': 'CDE2', 'batchCode': 'FGH3', 'ts': '2022/07/04 09:48:36'}
# 仅做入参时间格式合法性校验，不需要额外转ISO格式
datetime.strptime(payload['ts'], '%Y/%m/%d %H:%M:%S')

# 唯一键过滤条件
filter_rule = {
    'stCode': payload['stCode'],
    'skCode': payload['skCode'],
    'batchCode': payload['batchCode']
}

# 全在MongoDB侧执行的更新逻辑
update_pipeline = [
    {
        '$set': {
            '$cond': {
                'if': {
                    '$or': [
                        # 场景1：当前是upsert新建，原记录不存在
                        {'$eq': [{'$type': '$_id'}, 'missing']},
                        # 场景2：传入ts大于库内已有ts
                        {'$gt': [payload['ts'], '$ts']}
                    ]
                },
                # 满足条件直接替换为新传入的整条记录
                'then': payload,
                # 不满足条件保留原记录所有字段，不做任何修改
                'else': '$$ROOT'
            }
        }
    }
]

# 执行原子upsert
res = collection.update_one(
    filter=filter_rule,
    update=update_pipeline,
    upsert=True
)

# 按需判断执行结果
if res.upserted_id:
    print("新记录插入完成")
elif res.modified_count:
    print("旧记录更新完成（传入时间戳更新）")
else:
    print("无操作（传入时间戳早于库内记录）")

优化点说明 #

• 原子性：整个操作是MongoDB单文档原子操作，不会出现原代码「先find再update」间隙被其他写入覆盖的竞态问题，数据一致性更强
• 性能：仅需一次网络往返，比原写法少一次查询请求，延迟更低
• 可维护性：所有更新逻辑收敛在单条语句里，不需要维护Python侧的分支判断
• 扩展性：如果后续你把库内ts字段改成MongoDB原生BSON日期类型（查询性能更高、支持更多时间操作），只需要把payload里的ts替换为解析后的datetime对象即可，更新逻辑不需要改动

必做优化建议 #

给三个唯一字段建联合唯一索引，既可以从数据库层面兜底保证三个字段组合不重复，也能把upsert的查询速度提升几个数量级：

collection.create_index([("stCode", 1), ("skCode", 1), ("batchCode", 1)], unique=True)

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