Function _client_batched_op_msg_impl

pymongo/message.py:1041–1156

Create a batched OP_MSG write for client-level bulk write.

(
    command: Mapping[str, Any],
    operations: list[tuple[str, Mapping[str, Any]]],
    namespaces: list[str],
    ack: bool,
    opts: CodecOptions[Any],
    ctx: _ClientBulkWriteContext,
    buf: _BytesIO,
)

def _client_batched_op_msg_impl(
    command: Mapping[str, Any],
    operations: list[tuple[str, Mapping[str, Any]]],
    namespaces: list[str],
    ack: bool,
    opts: CodecOptions[Any],
    ctx: _ClientBulkWriteContext,
    buf: _BytesIO,
) -> tuple[list[Mapping[str, Any]], list[Mapping[str, Any]], int]:
    """Create a batched OP_MSG write for client-level bulk write."""

    def _check_doc_size_limits(
        op_type: str,
        doc_size: int,
        limit: int,
    ) -> None:
        if doc_size > limit:
            _raise_document_too_large(op_type, doc_size, limit)

    max_bson_size = ctx.max_bson_size
    max_write_batch_size = ctx.max_write_batch_size
    max_message_size = ctx.max_message_size

    command_encoded = _dict_to_bson(command, False, opts)
    # When OP_MSG is used unacknowledged we have to check command
    # document size client-side or applications won't be notified.
    if not ack:
        _check_doc_size_limits("bulkWrite", len(command_encoded), max_bson_size + _COMMAND_OVERHEAD)

    # Don't include bulkWrite-command-agnostic fields in batch-splitting calculations.
    abridged_keys = ["bulkWrite", "errorsOnly", "ordered"]
    if command.get("bypassDocumentValidation"):
        abridged_keys.append("bypassDocumentValidation")
    if command.get("comment"):
        abridged_keys.append("comment")
    if command.get("let"):
        abridged_keys.append("let")
    command_abridged = {key: command[key] for key in abridged_keys}
    command_len_abridged = len(_dict_to_bson(command_abridged, False, opts))

    # Maximum combined size of the ops and nsInfo document sequences.
    max_doc_sequences_bytes = max_message_size - (_OP_MSG_OVERHEAD + command_len_abridged)

    ns_info = {}
    to_send_ops: list[Mapping[str, Any]] = []
    to_send_ns: list[Mapping[str, str]] = []
    to_send_ops_encoded: list[bytes] = []
    to_send_ns_encoded: list[bytes] = []
    total_ops_length = 0
    total_ns_length = 0
    idx = 0

    for (real_op_type, op_doc), namespace in zip(operations, namespaces):
        op_type = real_op_type
        # Check insert/replace document size if unacknowledged.
        if real_op_type == "insert":
            if not ack:
                doc_size = len(_dict_to_bson(op_doc["document"], False, opts))