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