批量更新数据 Args: coll_name: 集合名 update_data_list: 更新数据列表 condition_field: 更新条件字段 upsert: 数据不存在则插入，默认为 False Returns: 更新行数
(
self,
coll_name: str,
update_data_list: List[Dict],
condition_field: str,
upsert: bool = False,
)
| 362 | return True |
| 363 | |
| 364 | def update_batch( |
| 365 | self, |
| 366 | coll_name: str, |
| 367 | update_data_list: List[Dict], |
| 368 | condition_field: str, |
| 369 | upsert: bool = False, |
| 370 | ): |
| 371 | """ |
| 372 | 批量更新数据 |
| 373 | Args: |
| 374 | coll_name: 集合名 |
| 375 | update_data_list: 更新数据列表 |
| 376 | condition_field: 更新条件字段 |
| 377 | upsert: 数据不存在则插入，默认为 False |
| 378 | |
| 379 | Returns: 更新行数 |
| 380 | |
| 381 | """ |
| 382 | if not update_data_list: |
| 383 | return 0 |
| 384 | |
| 385 | collection = self.get_collection(coll_name) |
| 386 | bulk_operations = [] |
| 387 | |
| 388 | for update_data in update_data_list: |
| 389 | condition = {condition_field: update_data.get(condition_field)} |
| 390 | update_operation = UpdateMany( |
| 391 | condition, {"$set": update_data}, upsert=upsert |
| 392 | ) |
| 393 | bulk_operations.append(update_operation) |
| 394 | try: |
| 395 | result = collection.bulk_write(bulk_operations, ordered=False) |
| 396 | return result.modified_count + result.upserted_count |
| 397 | except BulkWriteError as e: |
| 398 | log.error(f"Bulk write error: {e.details}") |
| 399 | return 0 |
| 400 | |
| 401 | def delete(self, coll_name, condition: Dict) -> bool: |
| 402 | """ |
no test coverage detected