import asyncio async def asyncgen(): yield 1 yield 2 async def main(): async for i in asyncgen(): print(i) asyncio.run(main())
I created a function which is able to wrap any asynchronous generator, the same way you would wrap a basic function using
def asyncgen_wrapper(generator): async def wrapped(*args, **kwargs): print("The wrapped asynchronous generator is iterated") gen = generator(*args, **kwargs) try: value = await gen.__anext__() except StopAsyncIteration: return while True: to_send = yield value try: value = await gen.asend(to_send) except StopAsyncIteration: return return wrapped
Wrapping an asynchronous generator seems quite complicated compared to wrapping a basic generator:
def gen_wrapper(generator): def wrapped(*args, **kwargs): return (yield from generator(*args, **kwargs)) return wrapped
I mainly concerned about correctness of my wrapper. I want the wrapper to be as transparent as possible regarding the wrapped generator. Which leads me to have two questions:
- Is there a more straightforward way to implement a decorator for asynchronous generators?
- Does my implementation handle all possible edge cases (think to