Python `beanie` mongo ODM：使用 upsert=True 进行批量更新答案

【问题标题】：Python `beanie` mongo ODM: Bulk update with upsert=TruePython `beanie` mongo ODM：使用 upsert=True 进行批量更新
【发布时间】：2022-07-04 09:58:58
【问题描述】：

我正在使用beanie==1.10.1

我想使用upsert=True 更新多个文档来执行批量操作。如果查找查询没有给出结果，我希望以下代码插入完整文档。我以此为参考：https://github.com/roman-right/beanie/blob/main/tests/odm/documents/test_bulk_write.py

这里是完整的代码：

import beanie
import asyncio
import random
from beanie import BulkWriter
from beanie.odm.operators.update.general import Set
from motor.motor_asyncio import AsyncIOMotorClient


class TestDoc(beanie.Document):
    a: str
    b: int


async def init_mongo():
    mongo_client = AsyncIOMotorClient("mongodb://127.0.0.1:27017")
    await beanie.init_beanie(
        database=mongo_client.db_name, document_models=[TestDoc]
    )


async def run_test():
    await init_mongo()

    docs = [TestDoc(a=f"id_{i}", b=random.randint(1, 100)) for i in range(10)]
    async with BulkWriter() as bulk_writer:
        for doc in docs:
            await TestDoc \
                .find_one({TestDoc.a: doc.a}, bulk_writer=bulk_writer) \
                .upsert(Set({TestDoc.b: doc.b}), on_insert=doc, bulk_writer=bulk_writer)
                # .update_one(Set(doc), bulk_writer=bulk_writer, upsert=True)

    read_docs = await TestDoc.find().to_list()
    print(f"read_docs: {read_docs}")


if __name__ == '__main__':
    pool = asyncio.get_event_loop()
    pool.run_until_complete(run_test())

执行后没有文档被插入到数据库中。不适用于 .upsert() 或 .update_one() 方法。实现该逻辑的正确方法是什么？

使用pymongo，这样的操作会这样写（并且有效）：

def write_reviews(self, docs: List[TestDoc]):
    operations = []
    for doc in docs:
        doc_dict = to_dict(doc)
        update_operation = pymongo.UpdateOne(
            {"a": doc.a}, {"$set": doc_dict}, upsert=True
        )
        operations.append(update_operation)

    result = self.test_collection.bulk_write(operations)

PS：这里无法创建beanie 标签。有人可以为我创建吗？

【问题讨论】：

标签： python mongodb python-asyncio tornado-motor

【解决方案1】：

这是旧的，你可能已经想通了，但自从它在谷歌搜索我的第一个结果后，我想我会回答的。

您当前使用 bulk_writer 的方式只是包装选项然后提交它们。

async def run_test():
    await init_mongo()

    docs = [TestDoc(a=f"id_{i}", b=random.randint(1, 100)) for i in range(10)]
    async with BulkWriter() as bulk_writer:
        for doc in docs:
            await TestDoc \
                .find_one({TestDoc.a: doc.a}) \
                .upsert(Set({TestDoc.b: doc.b}), on_insert=doc)
        bulk_writer.commit()        

    read_docs = await TestDoc.find().to_list()
    print(f"read_docs: {read_docs}")

Original Test in Beanie Test Suite

【讨论】：