Re:Show HN: Parallelise pipelines of Python async iterables
by cynik_ on Sunday September 06, 2020 @08:21AM
(#24390860)
Yeah, I was thinking along those lines. A more readable version of this library that I sketched out might be something like --
An eagerly executing coroutine to consume the original generator and put it in a queue, and a new simple generator that yields values from the queue till it's told to stop.
I'm not sure if this matches the behavior of the original library for task cancellation though.
def eager_gen(gen):
end_marker = object()
queue = asyncio.Queue()
async def consumer():
try:
async for val in gen:
await queue.put(val)
except Exception as e:
await queue.put((end_marker, e))
else:
await queue.put((end_marker, None))
async def new_gen():
while True:
item = await queue.get()
if isinstance(item, tuple) and item[0] == end_marker:
break
yield item
if item[1]:
raise item[1] from None
_task = asyncio.create_task(consumer())
return new_gen()
An eagerly executing coroutine to consume the original generator and put it in a queue, and a new simple generator that yields values from the queue till it's told to stop.
I'm not sure if this matches the behavior of the original library for task cancellation though.
[Edit] This would also make a good decorator.