(
self, operation: Union[_Query, _GetMore], conn: Connection, apply_timeout: bool = False
)
| 122 | self._monitor.request_check() |
| 123 | |
| 124 | def operation_to_command( |
| 125 | self, operation: Union[_Query, _GetMore], conn: Connection, apply_timeout: bool = False |
| 126 | ) -> tuple[dict[str, Any], str]: |
| 127 | cmd, db = operation.as_command(conn, apply_timeout) |
| 128 | # Support auto encryption |
| 129 | if operation.client._encrypter and not operation.client._encrypter._bypass_auto_encryption: |
| 130 | cmd = operation.client._encrypter.encrypt( # type: ignore[misc, assignment] |
| 131 | operation.db, cmd, operation.codec_options |
| 132 | ) |
| 133 | operation.update_command(cmd) |
| 134 | |
| 135 | return cmd, db |
| 136 | |
| 137 | @_handle_reauth |
| 138 | def run_operation( |
no test coverage detected