The following code works as expected:
import asyncio
from collections.abc import AsyncIterable, Iterable
# In reality, this is more complicated.
async def create_asynchronous_iterable() -> AsyncIterable[int]:
for k in range(3):
yield k
# In reality, this is more complicated.
def do_something(it: Iterable[int]) -> None:
for k in it:
print(k, end=" ")
print()
async def main():
async_it = create_asynchronous_iterable()
it = [k async for k in async_it]
do_something(it)
if __name__ == "__main__":
asyncio.run(main()) # Prints "0 1 2 ".
How can I turn it
into an iterable that avoids having all items in memory at once, e.g. a generator?
I tried to replace the declaration of main
as follows:
async def main():
async_it = create_asynchronous_iterable()
def create_iterable() -> Iterable[int]:
async for k in async_it:
yield k
it = create_iterable()
do_something(it)
However, this leads to:
SyntaxError: 'async for' outside async function
Is there a way to circumvent this limitation?
1