(
session: Optional[ClientSession],
_server: Server,
conn: Connection,
read_preference: _ServerMode,
)
| 933 | read_preference = (session and session._txn_read_preference()) or ReadPreference.PRIMARY |
| 934 | |
| 935 | def inner( |
| 936 | session: Optional[ClientSession], |
| 937 | _server: Server, |
| 938 | conn: Connection, |
| 939 | read_preference: _ServerMode, |
| 940 | ) -> Union[dict[str, Any], _CodecDocumentType]: |
| 941 | return self._command( |
| 942 | conn, |
| 943 | command, |
| 944 | value, |
| 945 | check, |
| 946 | allowable_errors, |
| 947 | read_preference, |
| 948 | opts, # type: ignore[arg-type] |
| 949 | session=session, |
| 950 | **kwargs, |
| 951 | ) |
| 952 | |
| 953 | return self._client._retryable_read( |
| 954 | inner, read_preference, session, command_name, None, False, is_run_command=True |
nothing calls this directly
no test coverage detected