(
self, cmd: MutableMapping[str, Any], docs: list[Mapping[str, Any]]
)
| 724 | __slots__ = () |
| 725 | |
| 726 | def batch_command( |
| 727 | self, cmd: MutableMapping[str, Any], docs: list[Mapping[str, Any]] |
| 728 | ) -> tuple[int, dict[str, Any], list[Mapping[str, Any]]]: |
| 729 | namespace = self.db_name + ".$cmd" |
| 730 | msg, to_send = _encode_batched_write_command( |
| 731 | namespace, self.op_type, cmd, docs, self.codec, self |
| 732 | ) |
| 733 | if not to_send: |
| 734 | raise InvalidOperation("cannot do an empty bulk write") |
| 735 | |
| 736 | # Chop off the OP_QUERY header to get a properly batched write command. |
| 737 | cmd_start = msg.index(b"\x00", 4) + 9 |
| 738 | outgoing = _inflate_bson(memoryview(msg)[cmd_start:], DEFAULT_RAW_BSON_OPTIONS) |
| 739 | return -1, outgoing, to_send |
| 740 | |
| 741 | @property |
| 742 | def max_split_size(self) -> int: |
nothing calls this directly
no test coverage detected