An :class:`_engine.IteratorResult` that works from an iterator-producing callable. The given ``chunks`` argument is a function that is given a number of rows to return in each chunk, or ``None`` for all rows. The function should then return an un-consumed iterator of lists, each li
| 2352 | |
| 2353 | |
| 2354 | class ChunkedIteratorResult(IteratorResult[_TP]): |
| 2355 | """An :class:`_engine.IteratorResult` that works from an |
| 2356 | iterator-producing callable. |
| 2357 | |
| 2358 | The given ``chunks`` argument is a function that is given a number of rows |
| 2359 | to return in each chunk, or ``None`` for all rows. The function should |
| 2360 | then return an un-consumed iterator of lists, each list of the requested |
| 2361 | size. |
| 2362 | |
| 2363 | The function can be called at any time again, in which case it should |
| 2364 | continue from the same result set but adjust the chunk size as given. |
| 2365 | |
| 2366 | .. versionadded:: 1.4 |
| 2367 | |
| 2368 | """ |
| 2369 | |
| 2370 | def __init__( |
| 2371 | self, |
| 2372 | cursor_metadata: ResultMetaData, |
| 2373 | chunks: Callable[ |
| 2374 | [Optional[int]], Iterator[Sequence[_InterimRowType[_R]]] |
| 2375 | ], |
| 2376 | source_supports_scalars: bool = False, |
| 2377 | raw: Optional[Result[Any]] = None, |
| 2378 | dynamic_yield_per: bool = False, |
| 2379 | ): |
| 2380 | self._metadata = cursor_metadata |
| 2381 | self.chunks = chunks |
| 2382 | self._source_supports_scalars = source_supports_scalars |
| 2383 | self.raw = raw |
| 2384 | self.iterator = itertools.chain.from_iterable(self.chunks(None)) |
| 2385 | self.dynamic_yield_per = dynamic_yield_per |
| 2386 | |
| 2387 | @_generative |
| 2388 | def yield_per(self, num: int) -> Self: |
| 2389 | # TODO: this throws away the iterator which may be holding |
| 2390 | # onto a chunk. the yield_per cannot be changed once any |
| 2391 | # rows have been fetched. either find a way to enforce this, |
| 2392 | # or we can't use itertools.chain and will instead have to |
| 2393 | # keep track. |
| 2394 | |
| 2395 | self._yield_per = num |
| 2396 | self.iterator = itertools.chain.from_iterable(self.chunks(num)) |
| 2397 | return self |
| 2398 | |
| 2399 | def _soft_close(self, hard: bool = False, **kw: Any) -> None: |
| 2400 | super()._soft_close(hard=hard, **kw) |
| 2401 | self.chunks = lambda size: [] # type: ignore |
| 2402 | |
| 2403 | def _fetchmany_impl( |
| 2404 | self, size: Optional[int] = None |
| 2405 | ) -> List[_InterimRowType[Row[Any]]]: |
| 2406 | if self.dynamic_yield_per: |
| 2407 | self.iterator = itertools.chain.from_iterable(self.chunks(size)) |
| 2408 | return super()._fetchmany_impl(size=size) |
| 2409 | |
| 2410 | |
| 2411 | class MergedResult(IteratorResult[_TP]): |