如何在pymongo中实现字典列表类型字段的更新或新增操作

你可以通过MongoDB 4.2+支持的更新聚合管道实现原子操作，无需先查询再修改，能避免并发冲突，完全适配你给出的场景。

前置说明 #

默认你的test_field数组中每个元素都是单key字典，输入的待操作字典也为单key结构，和示例一致。
我们假设待操作目标文档的_id为1，待写入的字典为input_dict。

方法1：原子更新聚合管道（推荐） #

MongoDB Shell 示例 #

// 定义待操作的字典
const inputDict = { "key1": "test_value" };
const inputKey = Object.keys(inputDict)[0];

db.collection.updateOne(
  { _id: 1 },
  [
    {
      $set: {
        test_field: {
          $cond: {
            // 第一步：判断test_field中是否已存在对应key
            if: {
              $in: [
                inputKey,
                { $map: { input: "$test_field", in: { $first: { $objectToArray: "$$this" } }.k } }
              ]
            },
            // 存在则遍历数组，更新对应key的字典
            then: {
              $map: {
                input: "$test_field",
                in: {
                  $cond: {
                    if: { $eq: [{ $first: { $objectToArray: "$$this" } }.k, inputKey] },
                    then: inputDict,
                    else: "$$this"
                  }
                }
              }
            },
            // 不存在则追加新字典到数组末尾
            else: { $concatArrays: [ "$test_field", [ inputDict ] ] }
          }
        }
      }
    }
  ]
)

Python PyMongo 示例 #

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
db = client.test_database
col = db.test_collection

# 定义待操作的字典
input_dict = {"test_key": "test_value2"}
input_key = next(iter(input_dict.keys()))

col.update_one(
    {"_id": 1},
    [
        {
            "$set": {
                "test_field": {
                    "$cond": {
                        "if": {
                            "$in": [
                                input_key,
                                {"$map": {
                                    "input": "$test_field",
                                    "in": {"$first": {"$objectToArray": "$$this"}}["k"]
                                }}
                            ]
                        },
                        "then": {
                            "$map": {
                                "input": "$test_field",
                                "in": {
                                    "$cond": {
                                        "if": {"$eq": [{"$first": {"$objectToArray": "$$this"}}["k"], input_key]},
                                        "then": input_dict,
                                        "else": "$$this"
                                    }
                                }
                            }
                        },
                        "else": {"$concatArrays": ["$test_field", [input_dict]]}
                    }
                }
            }
        }
    ]
)

方法2：低版本MongoDB兼容方案（先查后改） #

如果你的MongoDB版本低于4.2，不支持更新聚合管道，可以用先查询后修改的方式实现，注意如果有并发修改需求，需要加乐观锁或者事务保证数据一致性：

# 查询目标文档
doc = col.find_one({"_id": 1}, {"test_field": 1})
input_dict = {"key1": "test_value"}
input_key = next(iter(input_dict.keys()))

new_test_field = []
key_existed = False
for item in doc["test_field"]:
    item_key = next(iter(item.keys()))
    if item_key == input_key:
        new_test_field.append(input_dict)
        key_existed = True
    else:
        new_test_field.append(item)
#  key不存在则追加
if not key_existed:
    new_test_field.append(input_dict)

# 写回数据库
col.update_one({"_id": 1}, {"$set": {"test_field": new_test_field}})

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