Get all or some documents from the cursor.
(self, result: list, total: Optional[int] = None)
| 1147 | raise StopAsyncIteration |
| 1148 | |
| 1149 | async def _next_batch(self, result: list, total: Optional[int] = None) -> bool: # type: ignore[type-arg] |
| 1150 | """Get all or some documents from the cursor.""" |
| 1151 | if not self._exhaust_checked: |
| 1152 | self._exhaust_checked = True |
| 1153 | await self._supports_exhaust() |
| 1154 | if self._empty: |
| 1155 | return False |
| 1156 | if len(self._data) or await self._refresh(): |
| 1157 | if total is None: |
| 1158 | result.extend(self._data) |
| 1159 | self._data.clear() |
| 1160 | else: |
| 1161 | for _ in range(min(len(self._data), total)): |
| 1162 | result.append(self._data.popleft()) |
| 1163 | return True |
| 1164 | else: |
| 1165 | return False |
| 1166 | |
| 1167 | async def __anext__(self) -> _DocumentType: |
| 1168 | return await self.next() |
nothing calls this directly
no test coverage detected