(
self,
generator: Iterator[Any],
write_concern: WriteConcern,
session: Optional[ClientSession],
conn: Connection,
op_id: int,
retryable: bool,
full_result: MutableMapping[str, Any],
final_write_concern: Optional[WriteConcern] = None,
)
| 464 | return result, to_send # type: ignore[return-value] |
| 465 | |
| 466 | def _execute_command( |
| 467 | self, |
| 468 | generator: Iterator[Any], |
| 469 | write_concern: WriteConcern, |
| 470 | session: Optional[ClientSession], |
| 471 | conn: Connection, |
| 472 | op_id: int, |
| 473 | retryable: bool, |
| 474 | full_result: MutableMapping[str, Any], |
| 475 | final_write_concern: Optional[WriteConcern] = None, |
| 476 | ) -> None: |
| 477 | db_name = self.collection.database.name |
| 478 | client = self.collection.database.client |
| 479 | listeners = client._event_listeners |
| 480 | |
| 481 | if not self.current_run: |
| 482 | self.current_run = next(generator) |
| 483 | self.next_run = None |
| 484 | run = self.current_run |
| 485 | |
| 486 | # Connection.command validates the session, but we use |
| 487 | # Connection.write_command |
| 488 | conn.validate_session(client, session) |
| 489 | last_run = False |
| 490 | |
| 491 | while run: |
| 492 | if not self.retrying: |
| 493 | self.next_run = next(generator, None) |
| 494 | if self.next_run is None: |
| 495 | last_run = True |
| 496 | |
| 497 | cmd_name = _COMMANDS[run.op_type] |
| 498 | bwc = self.bulk_ctx_class( |
| 499 | db_name, |
| 500 | cmd_name, |
| 501 | conn, |
| 502 | op_id, |
| 503 | listeners, |
| 504 | session, |
| 505 | run.op_type, |
| 506 | self.collection.codec_options, |
| 507 | ) |
| 508 | |
| 509 | while run.idx_offset < len(run.ops): |
| 510 | # If this is the last possible operation, use the |
| 511 | # final write concern. |
| 512 | if last_run and (len(run.ops) - run.idx_offset) == 1: |
| 513 | write_concern = final_write_concern or write_concern |
| 514 | |
| 515 | cmd = {cmd_name: self.collection.name, "ordered": self.ordered} |
| 516 | if self.comment: |
| 517 | cmd["comment"] = self.comment |
| 518 | _csot.apply_write_concern(cmd, write_concern) |
| 519 | if self.bypass_doc_val is not None: |
| 520 | cmd["bypassDocumentValidation"] = self.bypass_doc_val |
| 521 | if self.let is not None and run.op_type in (_DELETE, _UPDATE): |
| 522 | cmd["let"] = self.let |
| 523 | if session: |
no test coverage detected