Run a _Query/_GetMore operation and return a Response. :param operation: a _Query or _GetMore object. :param unpack_res: A callable that decodes the wire protocol response. :param address: Optional address when sending a message to a specific server, used for get
(
self,
operation: Union[_Query, _GetMore],
unpack_res: Callable, # type: ignore[type-arg]
address: Optional[_Address] = None,
)
| 1905 | |
| 1906 | @_csot.apply |
| 1907 | def _run_operation( |
| 1908 | self, |
| 1909 | operation: Union[_Query, _GetMore], |
| 1910 | unpack_res: Callable, # type: ignore[type-arg] |
| 1911 | address: Optional[_Address] = None, |
| 1912 | ) -> Response: |
| 1913 | """Run a _Query/_GetMore operation and return a Response. |
| 1914 | |
| 1915 | :param operation: a _Query or _GetMore object. |
| 1916 | :param unpack_res: A callable that decodes the wire protocol response. |
| 1917 | :param address: Optional address when sending a message |
| 1918 | to a specific server, used for getMore. |
| 1919 | """ |
| 1920 | if operation.conn_mgr: |
| 1921 | server = self._select_server( |
| 1922 | operation.read_preference, |
| 1923 | operation.session, # type: ignore[arg-type] |
| 1924 | operation.name, |
| 1925 | address=address, |
| 1926 | ) |
| 1927 | |
| 1928 | with operation.conn_mgr._lock: |
| 1929 | with _MongoClientErrorHandler(self, server, operation.session) as err_handler: # type: ignore[arg-type] |
| 1930 | err_handler.contribute_socket(operation.conn_mgr.conn) |
| 1931 | return server.run_operation( |
| 1932 | operation.conn_mgr.conn, |
| 1933 | operation, |
| 1934 | operation.read_preference, |
| 1935 | self._event_listeners, |
| 1936 | unpack_res, |
| 1937 | self, |
| 1938 | ) |
| 1939 | |
| 1940 | def _cmd( |
| 1941 | _session: Optional[ClientSession], |
| 1942 | server: Server, |
| 1943 | conn: Connection, |
| 1944 | read_preference: _ServerMode, |
| 1945 | ) -> Response: |
| 1946 | operation.reset() # Reset op in case of retry. |
| 1947 | return server.run_operation( |
| 1948 | conn, |
| 1949 | operation, |
| 1950 | read_preference, |
| 1951 | self._event_listeners, |
| 1952 | unpack_res, |
| 1953 | self, |
| 1954 | ) |
| 1955 | |
| 1956 | return self._retryable_read( |
| 1957 | _cmd, |
| 1958 | operation.read_preference, |
| 1959 | operation.session, # type: ignore[arg-type] |
| 1960 | address=address, |
| 1961 | retryable=isinstance(operation, _Query), |
| 1962 | operation=operation.name, |
| 1963 | ) |
| 1964 |