Execute using write commands.
(
self,
generator: Iterator[Any],
write_concern: WriteConcern,
session: Optional[ClientSession],
operation: str,
)
| 567 | self.current_run = run = self.next_run |
| 568 | |
| 569 | def execute_command( |
| 570 | self, |
| 571 | generator: Iterator[Any], |
| 572 | write_concern: WriteConcern, |
| 573 | session: Optional[ClientSession], |
| 574 | operation: str, |
| 575 | ) -> dict[str, Any]: |
| 576 | """Execute using write commands.""" |
| 577 | # nModified is only reported for write commands, not legacy ops. |
| 578 | full_result = { |
| 579 | "writeErrors": [], |
| 580 | "writeConcernErrors": [], |
| 581 | "nInserted": 0, |
| 582 | "nUpserted": 0, |
| 583 | "nMatched": 0, |
| 584 | "nModified": 0, |
| 585 | "nRemoved": 0, |
| 586 | "upserted": [], |
| 587 | } |
| 588 | op_id = _randint() |
| 589 | |
| 590 | def retryable_bulk( |
| 591 | session: Optional[ClientSession], conn: Connection, retryable: bool |
| 592 | ) -> None: |
| 593 | self._execute_command( |
| 594 | generator, |
| 595 | write_concern, |
| 596 | session, |
| 597 | conn, |
| 598 | op_id, |
| 599 | retryable, |
| 600 | full_result, |
| 601 | ) |
| 602 | |
| 603 | client = self.collection.database.client |
| 604 | _ = client._retryable_write( |
| 605 | self.is_retryable, |
| 606 | retryable_bulk, |
| 607 | session, |
| 608 | operation, |
| 609 | bulk=self, # type: ignore[arg-type] |
| 610 | operation_id=op_id, |
| 611 | ) |
| 612 | |
| 613 | if full_result["writeErrors"] or full_result["writeConcernErrors"]: |
| 614 | _raise_bulk_write_error(full_result) |
| 615 | return full_result |
| 616 | |
| 617 | def execute_op_msg_no_results(self, conn: Connection, generator: Iterator[Any]) -> None: |
| 618 | """Execute write commands with OP_MSG and w=0 writeConcern, unordered.""" |
no test coverage detected