Internal listCollections helper.
(
self,
conn: Connection,
session: Optional[ClientSession],
read_preference: _ServerMode,
**kwargs: Any,
)
| 1087 | return self._client._retryable_read(_cmd, read_preference, session, operation) |
| 1088 | |
| 1089 | def _list_collections( |
| 1090 | self, |
| 1091 | conn: Connection, |
| 1092 | session: Optional[ClientSession], |
| 1093 | read_preference: _ServerMode, |
| 1094 | **kwargs: Any, |
| 1095 | ) -> CommandCursor[MutableMapping[str, Any]]: |
| 1096 | """Internal listCollections helper.""" |
| 1097 | coll = cast( |
| 1098 | Collection[MutableMapping[str, Any]], |
| 1099 | self.get_collection("$cmd", read_preference=read_preference), |
| 1100 | ) |
| 1101 | cmd = {"listCollections": 1, "cursor": {}} |
| 1102 | cmd.update(kwargs) |
| 1103 | with self._client._tmp_session(session) as tmp_session: |
| 1104 | cursor = ( |
| 1105 | self._command(conn, cmd, read_preference=read_preference, session=tmp_session) |
| 1106 | )["cursor"] |
| 1107 | cmd_cursor = CommandCursor( |
| 1108 | coll, |
| 1109 | cursor, |
| 1110 | conn.address, |
| 1111 | session=tmp_session, |
| 1112 | comment=cmd.get("comment"), |
| 1113 | ) |
| 1114 | cmd_cursor._maybe_pin_connection(conn) |
| 1115 | return cmd_cursor |
| 1116 | |
| 1117 | def _list_collections_helper( |
| 1118 | self, |
no test coverage detected