(
self,
command: _DocumentOut,
database_name: str,
request_id: int,
connection_id: _Address,
operation_id: Optional[int],
service_id: Optional[ObjectId] = None,
server_connection_id: Optional[int] = None,
)
| 614 | __slots__ = ("__cmd",) |
| 615 | |
| 616 | def __init__( |
| 617 | self, |
| 618 | command: _DocumentOut, |
| 619 | database_name: str, |
| 620 | request_id: int, |
| 621 | connection_id: _Address, |
| 622 | operation_id: Optional[int], |
| 623 | service_id: Optional[ObjectId] = None, |
| 624 | server_connection_id: Optional[int] = None, |
| 625 | ) -> None: |
| 626 | if not command: |
| 627 | raise ValueError(f"{command!r} is not a valid command") |
| 628 | # Command name must be first key. |
| 629 | command_name = next(iter(command)) |
| 630 | super().__init__( |
| 631 | command_name, |
| 632 | request_id, |
| 633 | connection_id, |
| 634 | operation_id, |
| 635 | service_id=service_id, |
| 636 | database_name=database_name, |
| 637 | server_connection_id=server_connection_id, |
| 638 | ) |
| 639 | cmd_name = command_name.lower() |
| 640 | if cmd_name in _SENSITIVE_COMMANDS or _is_speculative_authenticate(cmd_name, command): |
| 641 | self.__cmd: _DocumentOut = {} |
| 642 | else: |
| 643 | self.__cmd = command |
| 644 | |
| 645 | @property |
| 646 | def command(self) -> _DocumentOut: |
nothing calls this directly
no test coverage detected