(
session: Optional[ClientSession], conn: Connection, retryable_write: bool
)
| 810 | command["comment"] = comment |
| 811 | |
| 812 | def _insert_command( |
| 813 | session: Optional[ClientSession], conn: Connection, retryable_write: bool |
| 814 | ) -> None: |
| 815 | if bypass_doc_val is not None: |
| 816 | command["bypassDocumentValidation"] = bypass_doc_val |
| 817 | |
| 818 | result = conn.command( |
| 819 | self._database.name, |
| 820 | command, |
| 821 | write_concern=write_concern, |
| 822 | codec_options=self._write_response_codec_options, |
| 823 | session=session, |
| 824 | client=self._database.client, |
| 825 | retryable_write=retryable_write, |
| 826 | ) |
| 827 | |
| 828 | _check_write_command_response(result) |
| 829 | |
| 830 | self._database.client._retryable_write( |
| 831 | acknowledged, _insert_command, session, operation=_Op.INSERT |
nothing calls this directly
no test coverage detected