Managing exceptions while iterating asynchronously

Question

I am looking for a recommendation about how to manage a couple of asynchronous gRPC connections and handle exceptions.

My current workflow looks like this:

async for channel in get_some_channels(condition):
    try:
        GrpcStub(channel).CallAFunction()
    except AioRpcError:
        # handle exceptions

get_some_channels is an asynchronous generator and yields channels that fit the passed condition and also (re)opens channels in case they were not open before. In GrpcStub(channel).CallAFunction() things can go wrong though. A connection can be closed because the gRPC server choked on the request, a connection dropped or timed out. I'd really like to handle those issues in get_some_channels. This would make the use of the function leaner and I could mark this channel as broken or delete it. I could even try to 'fix' the issue and yield it again.

But this is just context. Basically my problem boils down to this example:

import asyncio

async def channels():
    for i in [True, True, False, True]:
        yield i

async def main():
    async for channel in channels():
        if not channel:
            raise Exception("Something went wrong")
        print("ping")

asyncio.run(main())

But as mentioned in this answer you cannot process exceptions raised in for loops of generators. So this example won't work.

What's a good practice to deal with exceptions while iterating lists asynchronously? I could add an asynchronous context manager like this

async for channel in channels():
    async with SafeContext(channel):
        # ....

but maybe there is a more elegant way.

Managing exceptions while iterating asynchronously

0 Answers