Execute operations.
(
self,
write_concern: WriteConcern,
session: Optional[AsyncClientSession],
operation: str,
)
| 725 | return await self.execute_op_msg_no_results(conn, generator) |
| 726 | |
| 727 | async def execute( |
| 728 | self, |
| 729 | write_concern: WriteConcern, |
| 730 | session: Optional[AsyncClientSession], |
| 731 | operation: str, |
| 732 | ) -> Any: |
| 733 | """Execute operations.""" |
| 734 | if not self.ops: |
| 735 | raise InvalidOperation("No operations to execute") |
| 736 | if self.executed: |
| 737 | raise InvalidOperation("Bulk operations can only be executed once.") |
| 738 | self.executed = True |
| 739 | write_concern = write_concern or self.collection.write_concern |
| 740 | session = _validate_session_write_concern(session, write_concern) |
| 741 | |
| 742 | if self.ordered: |
| 743 | generator = self.gen_ordered() |
| 744 | else: |
| 745 | generator = self.gen_unordered() |
| 746 | |
| 747 | client = self.collection.database.client |
| 748 | if not write_concern.acknowledged: |
| 749 | async with await client._conn_for_writes(session, operation) as connection: |
| 750 | await self.execute_no_results(connection, generator, write_concern) |
| 751 | return None |
| 752 | else: |
| 753 | return await self.execute_command(generator, write_concern, session, operation) |
no test coverage detected