Method execute_command

pymongo/asynchronous/bulk.py:569–615

Execute using write commands.

(
        self,
        generator: Iterator[Any],
        write_concern: WriteConcern,
        session: Optional[AsyncClientSession],
        operation: str,
    )

            self.current_run = run = self.next_run

    async def execute_command(
        self,
        generator: Iterator[Any],
        write_concern: WriteConcern,
        session: Optional[AsyncClientSession],
        operation: str,
    ) -> dict[str, Any]:
        """Execute using write commands."""
        # nModified is only reported for write commands, not legacy ops.
        full_result = {
            "writeErrors": [],
            "writeConcernErrors": [],
            "nInserted": 0,
            "nUpserted": 0,
            "nMatched": 0,
            "nModified": 0,
            "nRemoved": 0,
            "upserted": [],
        }
        op_id = _randint()

        async def retryable_bulk(
            session: Optional[AsyncClientSession], conn: AsyncConnection, retryable: bool
        ) -> None:
            await self._execute_command(
                generator,
                write_concern,
                session,
                conn,
                op_id,
                retryable,
                full_result,
            )

        client = self.collection.database.client
        _ = await client._retryable_write(
            self.is_retryable,
            retryable_bulk,
            session,
            operation,
            bulk=self,  # type: ignore[arg-type]
            operation_id=op_id,
        )

        if full_result["writeErrors"] or full_result["writeConcernErrors"]:
            _raise_bulk_write_error(full_result)
        return full_result

    async def execute_op_msg_no_results(
        self, conn: AsyncConnection, generator: Iterator[Any]