Internal helper for inserting a single document.
(
self,
doc: Mapping[str, Any],
ordered: bool,
write_concern: WriteConcern,
op_id: Optional[int],
bypass_doc_val: Optional[bool],
session: Optional[ClientSession],
comment: Optional[Any] = None,
)
| 793 | return BulkWriteResult({}, False) |
| 794 | |
| 795 | def _insert_one( |
| 796 | self, |
| 797 | doc: Mapping[str, Any], |
| 798 | ordered: bool, |
| 799 | write_concern: WriteConcern, |
| 800 | op_id: Optional[int], |
| 801 | bypass_doc_val: Optional[bool], |
| 802 | session: Optional[ClientSession], |
| 803 | comment: Optional[Any] = None, |
| 804 | ) -> Any: |
| 805 | """Internal helper for inserting a single document.""" |
| 806 | write_concern = write_concern or self.write_concern |
| 807 | acknowledged = write_concern.acknowledged |
| 808 | command = {"insert": self.name, "ordered": ordered, "documents": [doc]} |
| 809 | if comment is not None: |
| 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 |
| 832 | ) |
| 833 | |
| 834 | if not isinstance(doc, RawBSONDocument): |
| 835 | return doc.get("_id") |
| 836 | return None |
| 837 | |
| 838 | def insert_one( |
| 839 | self, |
no test coverage detected