What Is The Correct Way To Combine Async-for With An If Condition To Break Mid Await?

If I have a coroutine that's consuming items from an async generator what is the 'best' way to terminate that loop from an external condition? Consider this, while not self.shutdow

Solution 1:

One way would be to move the iteration to an async def and use cancelation:

asyncdefiterate(client):
    asyncfor message in client:
        # shield() because we want cancelation to cancel retrieval# of the next message, not ongoing handling of a messageawait asyncio.shield(self.handle(message))

asyncwith self.external_lib_client as client:
    iter_task = asyncio.create_task(iterate(client))
    shutdown_task = asyncio.create_task(self.shutdown_event.wait())
    await asyncio.wait([iter_task, shutdown_task],
                       return_when=asyncio.FIRST_COMPLETED)
    if iter_task.done():
        # iteration has completed, access result to propagate the# exception if one was raised
        iter_task.result()
        shutdown_task.cancel()
    else:
        # shutdown was requested, cancel iteration
        iter_task.cancel()

Another way would be to turn shutdown_event into a one-shot async stream and use aiostream to monitor both. That way the for loop gets an object when the shutdown event is signaled and can break out of the loop without bothering to finish waiting for the next message:

# a stream that just yields something (the return value of `wait()`)# when shutdown_event is set
done_stream = aiostream.stream.just(self.shutdown_event.wait())

asyncwith self.external_lib_client as client, \
        aiostream.stream.merge(done_stream, client).stream() as stream:
    asyncfor message in stream:
        # the merged stream will provide a bogus value (whatever# `shutdown_event.wait()` returned) when the event is set,# so check that before using `message`:if self.shutdown_event.is_set():
            breakawait self.handle(message)

Note: since the code in the question is not runnable, the above examples are untested.