Something close to gather_cm can be achieved with AsyncExitStack, introduced in Python 3.7:
async def foo():
async with AsyncExitStack() as stack:
vm1, vm2 = await asyncio.gather(
stack.enter_async_context(start_vm()),
stack.enter_async_context(start_vm()))
await vm1.do_something()
await vm2.do_something()
Unfortunately, __aexit__s will still be run sequentially. This is because AsyncExitStack simulates nested context managers, which have a well-defined order and cannot overlap. The outer context manager's __aexit__ is given information on whether the inner one raised an exception. (A database handle's __aexit__ might use this to roll back the transaction in case of exception and commit it otherwise.) Running __aexit__s in parallel would make the context managers overlap and the exception information unavailable or unreliable. So although gather(...) runs __aenter__s in parallel, AsyncExitStack records which one came first and runs the __aexit__s in reverse order.
With async context managers an alternative like gather_cm would make perfect sense. One could drop the nesting semantics and provide an aggregate context manager that worked like an "exit pool" rather than a stack. The exit pool takes a number of context manager that are independent of each other, which allows their __aenter__ and __aexit__ methods to be run in parallel.
The tricky part is handling exceptions correctly: If any __aenter__ raises, the exception must be propagated to prevent the with block from being run. To ensure correctness, the pool must guarantee that __aexit__ will be invoked on all the context managers whose __aenter__ has completed.
Here is an example implementation:
import asyncio
import sys
class gather_cm:
def __init__(self, *cms):
self._cms = cms
async def __aenter__(self):
futs = [asyncio.create_task(cm.__aenter__())
for cm in self._cms]
await asyncio.wait(futs)
# only exit the cms we've successfully entered
self._cms = [cm for cm, fut in zip(self._cms, futs)
if not fut.cancelled() and not fut.exception()]
try:
return tuple(fut.result() for fut in futs)
except:
await self._exit(*sys.exc_info())
raise
async def _exit(self, *args):
# don't use gather() to ensure that we wait for all __aexit__s
# to complete even if one of them raises
done, _pending = await asyncio.wait(
[cm.__aexit__(*args)
for cm in self._cms if cm is not None])
return all(suppress.result() for suppress in done)
async def __aexit__(self, *args):
# Since exits are running in parallel, so they can't see each
# other exceptions. Send exception info from `async with`
# body to all.
return await self._exit(*args)
This test program shows how it works:
class test_cm:
def __init__(self, x):
self.x = x
async def __aenter__(self):
print('__aenter__', self.x)
return self.x
async def __aexit__(self, *args):
print('__aexit__', self.x, args)
async def foo():
async with gather_cm(test_cm('foo'), test_cm('bar')) as (cm1, cm2):
print('cm1', cm1)
print('cm2', cm2)
asyncio.run(foo())