(
self, cmd: MutableMapping[str, Any], docs: list[Mapping[str, Any]]
)
| 693 | ) |
| 694 | |
| 695 | def batch_command( |
| 696 | self, cmd: MutableMapping[str, Any], docs: list[Mapping[str, Any]] |
| 697 | ) -> tuple[int, Union[bytes, dict[str, Any]], list[Mapping[str, Any]]]: |
| 698 | namespace = self.db_name + ".$cmd" |
| 699 | request_id, msg, to_send = _do_batched_op_msg( |
| 700 | namespace, self.op_type, cmd, docs, self.codec, self |
| 701 | ) |
| 702 | if not to_send: |
| 703 | raise InvalidOperation("cannot do an empty bulk write") |
| 704 | return request_id, msg, to_send |
| 705 | |
| 706 | def _start( |
| 707 | self, cmd: MutableMapping[str, Any], request_id: int, docs: list[Mapping[str, Any]] |
no test coverage detected