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