Send a query or getmore operation and handles the response. If operation is ``None`` this is an exhaust cursor, which reads the next result batch off the exhaust socket instead of sending getMore messages to the server. Can raise ConnectionFailure.
(self, operation: Union[_Query, _GetMore])
| 968 | return self._collection.distinct(key, session=self._session, **options) |
| 969 | |
| 970 | def _send_message(self, operation: Union[_Query, _GetMore]) -> None: |
| 971 | """Send a query or getmore operation and handles the response. |
| 972 | |
| 973 | If operation is ``None`` this is an exhaust cursor, which reads |
| 974 | the next result batch off the exhaust socket instead of |
| 975 | sending getMore messages to the server. |
| 976 | |
| 977 | Can raise ConnectionFailure. |
| 978 | """ |
| 979 | client = self._collection.database.client |
| 980 | # OP_MSG is required to support exhaust cursors with encryption. |
| 981 | if client._encrypter and self._exhaust: |
| 982 | raise InvalidOperation("exhaust cursors do not support auto encryption") |
| 983 | |
| 984 | try: |
| 985 | response = client._run_operation( |
| 986 | operation, self._unpack_response, address=self._address |
| 987 | ) |
| 988 | except OperationFailure as exc: |
| 989 | if exc.code in _CURSOR_CLOSED_ERRORS or self._exhaust: |
| 990 | # Don't send killCursors because the cursor is already closed. |
| 991 | self._killed = True |
| 992 | if exc.timeout: |
| 993 | self._die_no_lock() |
| 994 | else: |
| 995 | self.close() |
| 996 | # If this is a tailable cursor the error is likely |
| 997 | # due to capped collection roll over. Setting |
| 998 | # self._killed to True ensures Cursor.alive will be |
| 999 | # False. No need to re-raise. |
| 1000 | if ( |
| 1001 | exc.code in _CURSOR_CLOSED_ERRORS |
| 1002 | and self._query_flags & _QUERY_OPTIONS["tailable_cursor"] |
| 1003 | ): |
| 1004 | return |
| 1005 | raise |
| 1006 | except ConnectionFailure: |
| 1007 | self._killed = True |
| 1008 | self.close() |
| 1009 | raise |
| 1010 | # Catch KeyboardInterrupt, CancelledError, etc. and cleanup. |
| 1011 | except BaseException: |
| 1012 | self.close() |
| 1013 | raise |
| 1014 | self._address = response.address |
| 1015 | if isinstance(response, PinnedResponse): |
| 1016 | if not self._sock_mgr: |
| 1017 | self._sock_mgr = _ConnectionManager(response.conn, response.more_to_come) # type: ignore[arg-type] |
| 1018 | |
| 1019 | cmd_name = operation.name |
| 1020 | docs = response.docs |
| 1021 | if response.from_command: |
| 1022 | if cmd_name != "explain": |
| 1023 | cursor = docs[0]["cursor"] |
| 1024 | self._id = cursor["id"] |
| 1025 | if cmd_name == "find": |
| 1026 | documents = cursor["firstBatch"] |
| 1027 | # Update the namespace used for future getMore commands. |
no test coverage detected