(
self,
duration: datetime.timedelta,
reply: _DocumentOut,
command_name: str,
request_id: int,
connection_id: _Address,
operation_id: Optional[int],
service_id: Optional[ObjectId] = None,
database_name: str = "",
server_connection_id: Optional[int] = None,
)
| 683 | __slots__ = ("__duration_micros", "__reply") |
| 684 | |
| 685 | def __init__( |
| 686 | self, |
| 687 | duration: datetime.timedelta, |
| 688 | reply: _DocumentOut, |
| 689 | command_name: str, |
| 690 | request_id: int, |
| 691 | connection_id: _Address, |
| 692 | operation_id: Optional[int], |
| 693 | service_id: Optional[ObjectId] = None, |
| 694 | database_name: str = "", |
| 695 | server_connection_id: Optional[int] = None, |
| 696 | ) -> None: |
| 697 | super().__init__( |
| 698 | command_name, |
| 699 | request_id, |
| 700 | connection_id, |
| 701 | operation_id, |
| 702 | service_id=service_id, |
| 703 | database_name=database_name, |
| 704 | server_connection_id=server_connection_id, |
| 705 | ) |
| 706 | self.__duration_micros = _to_micros(duration) |
| 707 | cmd_name = command_name.lower() |
| 708 | if cmd_name in _SENSITIVE_COMMANDS or _is_speculative_authenticate(cmd_name, reply): |
| 709 | self.__reply: _DocumentOut = {} |
| 710 | else: |
| 711 | self.__reply = reply |
| 712 | |
| 713 | @property |
| 714 | def duration_micros(self) -> int: |
nothing calls this directly
no test coverage detected