Create a batched OP_MSG write.
(
operation: int,
command: Mapping[str, Any],
docs: list[Mapping[str, Any]],
ack: bool,
opts: CodecOptions[Any],
ctx: _BulkWriteContext,
buf: _BytesIO,
)
| 767 | |
| 768 | |
| 769 | def _batched_op_msg_impl( |
| 770 | operation: int, |
| 771 | command: Mapping[str, Any], |
| 772 | docs: list[Mapping[str, Any]], |
| 773 | ack: bool, |
| 774 | opts: CodecOptions[Any], |
| 775 | ctx: _BulkWriteContext, |
| 776 | buf: _BytesIO, |
| 777 | ) -> tuple[list[Mapping[str, Any]], int]: |
| 778 | """Create a batched OP_MSG write.""" |
| 779 | max_bson_size = ctx.max_bson_size |
| 780 | max_write_batch_size = ctx.max_write_batch_size |
| 781 | max_message_size = ctx.max_message_size |
| 782 | |
| 783 | flags = b"\x00\x00\x00\x00" if ack else b"\x02\x00\x00\x00" |
| 784 | # Flags |
| 785 | buf.write(flags) |
| 786 | |
| 787 | # Type 0 Section |
| 788 | buf.write(b"\x00") |
| 789 | buf.write(_dict_to_bson(command, False, opts)) |
| 790 | |
| 791 | # Type 1 Section |
| 792 | buf.write(b"\x01") |
| 793 | size_location = buf.tell() |
| 794 | # Save space for size |
| 795 | buf.write(b"\x00\x00\x00\x00") |
| 796 | try: |
| 797 | buf.write(_OP_MSG_MAP[operation]) |
| 798 | except KeyError: |
| 799 | raise InvalidOperation("Unknown command") from None |
| 800 | |
| 801 | to_send = [] |
| 802 | idx = 0 |
| 803 | for doc in docs: |
| 804 | # Encode the current operation |
| 805 | value = _dict_to_bson(doc, False, opts) |
| 806 | doc_length = len(value) |
| 807 | new_message_size = buf.tell() + doc_length |
| 808 | # Does first document exceed max_message_size? |
| 809 | doc_too_large = idx == 0 and (new_message_size > max_message_size) |
| 810 | # When OP_MSG is used unacknowledged we have to check |
| 811 | # document size client side or applications won't be notified. |
| 812 | # Otherwise we let the server deal with documents that are too large |
| 813 | # since ordered=False causes those documents to be skipped instead of |
| 814 | # halting the bulk write operation. |
| 815 | unacked_doc_too_large = not ack and (doc_length > max_bson_size) |
| 816 | if doc_too_large or unacked_doc_too_large: |
| 817 | write_op = list(_FIELD_MAP.keys())[operation] |
| 818 | _raise_document_too_large(write_op, len(value), max_bson_size) |
| 819 | # We have enough data, return this batch. |
| 820 | if new_message_size > max_message_size: |
| 821 | break |
| 822 | buf.write(value) |
| 823 | to_send.append(doc) |
| 824 | idx += 1 |
| 825 | # We have enough documents, return this batch. |
| 826 | if idx == max_write_batch_size: |
no test coverage detected