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[AsyncClientSession],
comment: Optional[Any] = None,
)
| 794 | return BulkWriteResult({}, False) |
| 795 | |
| 796 | async def _insert_one( |
| 797 | self, |
| 798 | doc: Mapping[str, Any], |
| 799 | ordered: bool, |
| 800 | write_concern: WriteConcern, |
| 801 | op_id: Optional[int], |
| 802 | bypass_doc_val: Optional[bool], |
| 803 | session: Optional[AsyncClientSession], |
| 804 | comment: Optional[Any] = None, |
| 805 | ) -> Any: |
| 806 | """Internal helper for inserting a single document.""" |
| 807 | write_concern = write_concern or self.write_concern |
| 808 | acknowledged = write_concern.acknowledged |
| 809 | command = {"insert": self.name, "ordered": ordered, "documents": [doc]} |
| 810 | if comment is not None: |
| 811 | command["comment"] = comment |
| 812 | |
| 813 | async def _insert_command( |
| 814 | session: Optional[AsyncClientSession], conn: AsyncConnection, retryable_write: bool |
| 815 | ) -> None: |
| 816 | if bypass_doc_val is not None: |
| 817 | command["bypassDocumentValidation"] = bypass_doc_val |
| 818 | |
| 819 | result = await conn.command( |
| 820 | self._database.name, |
| 821 | command, |
| 822 | write_concern=write_concern, |
| 823 | codec_options=self._write_response_codec_options, |
| 824 | session=session, |
| 825 | client=self._database.client, |
| 826 | retryable_write=retryable_write, |
| 827 | ) |
| 828 | |
| 829 | _check_write_command_response(result) |
| 830 | |
| 831 | await self._database.client._retryable_write( |
| 832 | acknowledged, _insert_command, session, operation=_Op.INSERT |
| 833 | ) |
| 834 | |
| 835 | if not isinstance(doc, RawBSONDocument): |
| 836 | return doc.get("_id") |
| 837 | return None |
| 838 | |
| 839 | async def insert_one( |
| 840 | self, |
no test coverage detected