Send "insert" etc. command, returning response as a dict. Can raise ConnectionFailure or OperationFailure. :param request_id: an int. :param msg: bytes, the command message.
(
self, request_id: int, msg: bytes, codec_options: CodecOptions[Mapping[str, Any]]
)
| 478 | self.send_message(msg, max_doc_size) |
| 479 | |
| 480 | def write_command( |
| 481 | self, request_id: int, msg: bytes, codec_options: CodecOptions[Mapping[str, Any]] |
| 482 | ) -> dict[str, Any]: |
| 483 | """Send "insert" etc. command, returning response as a dict. |
| 484 | |
| 485 | Can raise ConnectionFailure or OperationFailure. |
| 486 | |
| 487 | :param request_id: an int. |
| 488 | :param msg: bytes, the command message. |
| 489 | """ |
| 490 | self.send_message(msg, 0) |
| 491 | reply = self.receive_message(request_id) |
| 492 | result = reply.command_response(codec_options) |
| 493 | |
| 494 | # Raises NotPrimaryError or OperationFailure. |
| 495 | helpers_shared._check_command_response(result, self.max_wire_version) |
| 496 | return result |
| 497 | |
| 498 | def authenticate(self, reauthenticate: bool = False) -> None: |
| 499 | """Authenticate to the server if needed. |
nothing calls this directly
no test coverage detected