Python ` `beanie`‘mongo ODM:使用upsert=True进行批量更新
Python ` `beanie`‘mongo ODM:使用upsert=True进行批量更新
提问于 2022-03-21 18:00:46
我正在使用beanie==1.10.1
我想通过使用upsert=True更新多个文档来执行批量操作。如果find查询没有给出结果，我希望下面的代码插入完整的文档。我用这个作为参考：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())执行后，不将任何文档插入到db中。不使用.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标记。有人能为我创造它吗？
回答 1
Stack Overflow用户
发布于 2022-07-04 01:58:58
这是个古老的问题，你可能已经想出了答案，但因为这是谷歌搜索的第一个结果，我想我会回答的。
当前使用bulk_writer的方式只是包装选项，然后提交它们。
from beanie.odm.operators.update.general import Set
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}")页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/71562082
复制相关文章
相似问题
腾讯云开发者
