Method unack_write

pymongo/synchronous/client_bulk.py:320–407

A proxy for Connection.unack_write that handles event publishing.

(
        self,
        bwc: _ClientBulkWriteContext,
        cmd: MutableMapping[str, Any],
        request_id: int,
        msg: bytes,
        op_docs: list[Mapping[str, Any]],
        ns_docs: list[Mapping[str, Any]],
        client: MongoClient[Any],
    )

        return reply  # type: ignore[return-value]

    def unack_write(
        self,
        bwc: _ClientBulkWriteContext,
        cmd: MutableMapping[str, Any],
        request_id: int,
        msg: bytes,
        op_docs: list[Mapping[str, Any]],
        ns_docs: list[Mapping[str, Any]],
        client: MongoClient[Any],
    ) -> Optional[Mapping[str, Any]]:
        """A proxy for Connection.unack_write that handles event publishing."""
        if _COMMAND_LOGGER.isEnabledFor(logging.DEBUG):
            _debug_log(
                _COMMAND_LOGGER,
                message=_CommandStatusMessage.STARTED,
                clientId=client._topology_settings._topology_id,
                command=cmd,
                commandName=next(iter(cmd)),
                databaseName=bwc.db_name,
                requestId=request_id,
                operationId=request_id,
                driverConnectionId=bwc.conn.id,
                serverConnectionId=bwc.conn.server_connection_id,
                serverHost=bwc.conn.address[0],
                serverPort=bwc.conn.address[1],
                serviceId=bwc.conn.service_id,
            )
        if bwc.publish:
            cmd = bwc._start(cmd, request_id, op_docs, ns_docs)
        try:
            result = bwc.conn.unack_write(msg, bwc.max_bson_size)  # type: ignore[func-returns-value, misc, override]
            duration = datetime.datetime.now() - bwc.start_time
            if result is not None:
                reply = _convert_write_result(bwc.name, cmd, result)  # type: ignore[arg-type]
            else:
                # Comply with APM spec.
                reply = {"ok": 1}
                if _COMMAND_LOGGER.isEnabledFor(logging.DEBUG):
                    _debug_log(
                        _COMMAND_LOGGER,
                        message=_CommandStatusMessage.SUCCEEDED,
                        clientId=client._topology_settings._topology_id,
                        durationMS=duration,
                        reply=reply,
                        commandName=next(iter(cmd)),
                        databaseName=bwc.db_name,
                        requestId=request_id,
                        operationId=request_id,
                        driverConnectionId=bwc.conn.id,
                        serverConnectionId=bwc.conn.server_connection_id,
                        serverHost=bwc.conn.address[0],
                        serverPort=bwc.conn.address[1],
                        serviceId=bwc.conn.service_id,
                    )
            if bwc.publish:
                bwc._succeed(request_id, reply, duration)
        except Exception as exc:
            duration = datetime.datetime.now() - bwc.start_time