将记录从一个集合移动到另一个 PyMongo答案

【问题标题】：Moving records from one collection to another PyMongo将记录从一个集合移动到另一个 PyMongo
【发布时间】：2017-12-04 22:04:06
【问题描述】：

将多个记录从一个集合移动到另一个集合的正确方法是什么。我遇到过其他几个 SO 帖子，例如 this，它们处理实现相同的目标，但没有一个有 python 实现。

#Taking a number of records from one database and returning the cursor
cursor_excess_new = db.test_collection_new.find().sort([("_id", 1)]).limit(excess_num)


# db.test.insert_many(doc for doc in cursor_excess_new).inserted_ids


# Iterating cursor and Trying to write to another database
# for doc in cursor_excess_new:
#     db.test_collection_old.insert_one(doc)
result = db.test_collection_old.bulk_write([
    for doc in cursor_excess_new:
        InsertMany(doc for each doc in cur)
        pprint(doc)
])

如果我使用 insert_many，我会收到以下错误：pymongo.errors.OperationFailure: Writes to config servers must have batch size of 1, found 10

bulk_write 在 for 循环开始时给我一个语法错误。

在 pymongo 中将记录从一个集合传输到另一个集合以使其具有原子性的最佳实践和正确方法是什么？

【问题讨论】：

标签： python database mongodb nosql pymongo

【解决方案1】：

Collection.bulk_write 接受一个可迭代的查询操作作为参数。

pymongo 有 pymongo.operations.InsertOne 操作不是 ~~InsertMany~~。

根据您的情况，您可以为源集合中的每个文档构建一个InsertOne 操作列表。然后使用构建的操作列表在目标上执行 bulk_write。

from pymongo import InsertOne
...
cursor_excess_new = (
    db.test_collection_new
      .find()
      .sort([("_id", 1)])
      .limit(excess_num)
)

queries = [InsertOne(doc) for doc in cursor_excess_new]
db.test_collection_old.bulk_write(queries)

【讨论】：

【解决方案2】：

你不需要“for循环”。

   myList=list(collection1.find({}))
   collection2.insert_many(myList)
   collection1.delete_many({})

如果需要过滤，可以使用以下代码：

   myList=list(collection1.find({'status':10}))
   collection2.insert_many(myList)
   collection1.delete_many({'status':10})

但是要小心，因为它没有保证移动成功，所以你需要控制交易。如果您要使用以下代码，您应该考虑到您的 MongoDb 不应该是独立的，您需要激活复制并拥有另一个实例。

  with myClient.start_session() as mySession:
      with mySession.start_transaction():
          ...yourcode...

最后，上面的代码有保证移动（插入和删除）成功，但交易不在你手上，你不能得到这个交易的结果，所以你可以使用下面的代码来控制移动和交易：

  with myClient.start_session() as mySession:
      mySession.start_transaction()
      try:
          ...yourcode...
          mySession.commit_transaction()
          print("Done")
      except Exception as e:
          mySession.abort_transaction()
          print("Failed",e)

【讨论】：