Well, here are two full program examples including the correct type information that would correspond with each. I’ll try and think about how the docs could be improved to reflect what’s actually needed later, as I can see how the current examples don’t quite do enough here.
example program implementing this with re-usability
import asyncio
from collections.abc import AsyncIterator, Awaitable, Callable
from typing import Generic, TypeVar
T = TypeVar("T")
class X(Generic[T]):
"""
Parameters
----------
fill_func: Callable[[int, int], Awaitable[list[T]]]
A callable to fill in values. should take two values, the first being
the first element needed, the second being the last element needed, inclusive, 0-indexed.
If the start is out of bounds, return an empty list
If the end is out of bounds but the start is not, return the remaining elements
eg: async def filler(start: int, stop: int) -> list:
"""
def __init__(self, fill_func: Callable[[int, int], Awaitable[list[T]]]) -> None:
self._values: list[T] = []
self._fill_func: Callable[[int, int], Awaitable[list[T]]] = fill_func
self._exhausted: bool = False
async def _get_value(self, index: int) -> T:
try:
return self._values[index]
except IndexError:
if self._exhausted:
raise
vals = await self._fill_func(len(self._values), index)
if not vals:
self._exhausted = True
raise
self._values.extend(vals)
return self._values[index]
def __aiter__(self):
async def iterator() -> AsyncIterator[T]:
index = 0
while index < len(self._values) or not self._exhausted:
try:
yield (await self._get_value(index))
except IndexError:
return
index += 1
return iterator()
async def filler(start: int, stop: int) -> list[int]:
""" This here to simulate some external paginated request API """
if start > 10:
return []
print(f"filling from {start} to {stop} inclusive")
stop = min(11, stop + 1)
return list(range(start, stop))
async def main() -> None:
x = X(filler)
async for i in x:
print(i)
async for i in x:
print(i)
if __name__ == "__main__":
asyncio.run(main())
The output of this is:
filling from 0 to 0 inclusive
0
filling from 1 to 1 inclusive
1
filling from 2 to 2 inclusive
2
filling from 3 to 3 inclusive
3
filling from 4 to 4 inclusive
4
filling from 5 to 5 inclusive
5
filling from 6 to 6 inclusive
6
filling from 7 to 7 inclusive
7
filling from 8 to 8 inclusive
8
filling from 9 to 9 inclusive
9
filling from 10 to 10 inclusive
10
0
1
2
3
4
5
6
7
8
9
10
And the equivalent for implementing it on the class itself without reusability
import asyncio
from collections.abc import Awaitable, Callable
from typing import Generic, TypeVar
T = TypeVar("T")
class X(Generic[T]):
"""
Parameters
----------
fill_func: Callable[[int, int], Awaitable[list[T]]]
A callable to fill in values. should take two values, the first being
the first element needed, the second being the last element needed, inclusive, 0-indexed.
If the start is out of bounds, return an empty list
If the end is out of bounds but the start is not, return the remaining elements
eg: async def filler(start: int, stop: int) -> list:
"""
def __init__(self, fill_func: Callable[[int, int], Awaitable[list[T]]]) -> None:
self._lock = asyncio.Lock()
self._fill_func: Callable[[int, int], Awaitable[list[T]]] = fill_func
self._exhausted: bool = False
self._current_index: int = 0
async def __anext__(self) -> T:
async with self._lock:
try:
if self._exhausted:
raise StopAsyncIteration from None
vals = await self._fill_func(self._current_index, self._current_index)
if not vals:
self._exhausted = True
raise StopAsyncIteration from None
return vals[0]
finally:
self._current_index += 1
def __aiter__(self):
return self
async def filler(start: int, stop: int) -> list[int]:
if start > 10:
return []
print(f"filling from {start} to {stop} inclusive")
stop = min(11, stop + 1)
return list(range(start, stop))
async def main() -> None:
x = X(filler)
async for i in x:
print(i)
async for i in x:
print(i) # Note: this will never print
if __name__ == "__main__":
asyncio.run(main())
Each pattern has merits, which one is correct for your own use will depend on your desired API. Depending on the actual use, you might want to fetch more than just the next value needed (such as the next n based on a paginated API). This also includes any necessary synchronization.