Interface

AioPool(size=4, *, loop=None)

spawn(coro, cb=None, ctx=None)

exec(coro, cb=None, ctx=None)

spawn_n(coro, cb=None, ctx=None)

join()

cancel(*futures)

map(fn, iterable, cb=None, ctx=None, *, get_result=getres.flat)

map_n(fn, iterable, cb=None, ctx=None, *, get_result=getres.flat)

itermap(fn, iterable, cb=None, ctx=None, *, flat=True, get_result=getres.flat, timeout=None, yield_when=asyncio.ALL_COMPLETED)

Usage

asyncdefworker(n):# dummy workerawaitaio.sleep(1/n)returnnasyncdefspawn_n_usage(todo=[range(1,51),range(51,101),range(101,200)]):futures= []asyncwithAioPool(size=20)aspool:fortasksintodo:foriintasks:# too many tasks# Returns quickly for all tasks, does not wait for pool space.# Workers are not spawned, they wait for pool space in their# own background tasks.fut=pool.spawn_n(worker(i))futures.append(fut)# At this point not a single worker should start.# Context manager calls `join` at exit, so this will finish when all# workers return, crash or cancelled.assertsum(itertools.chain.from_iterable(todo))== \sum(f.result()forfinfutures)asyncdefspawn_usage(todo=range(1,4)):futures= []asyncwithAioPool(size=2)aspool:foriintodo:# 1, 2, 3# Returns quickly for 1 and 2, then waits for empty space for 3,# spawns 3 and returns. Can save some resources I guess.fut=awaitpool.spawn(worker(i))futures.append(fut)# At this point some of the workers already started.# Context manager calls `join` at exit, so this will finish when all# workers return, crash or cancelled.assertsum(todo)==sum(fut.result()forfutinfutures)# all doneasyncdefmap_usage(todo=range(100)):pool=AioPool(size=10)# Waits and collects results from all spawned workers,# returns them in same order as `todo`, if worker crashes or cancelled:# returns exception object as a result.# Basically, it wraps `spawn_usage` code into one call.results=awaitpool.map(worker,todo)# await pool.join()  # is not needed here, bcs no other tasks were spawnedassertisinstance(results[0],ZeroDivisionError) \andsum(results[1:])==sum(todo)asyncdefitermap_usage(todo=range(1,11)):result=0asyncwithAioPool(size=10)aspool:# Combines spawn_n and iterwait, which is a wrapper for asyncio.wait,# which yields results of finished workers according to `timeout` and# `yield_when` params passed to asyncio.wait (see it's docs for details)asyncforresinpool.itermap(worker,todo,timeout=0.5):result+=res# technically, you can skip join callassertresult==sum(todo)asyncdefcallbacks_usage():asyncdefwrk(n):# custom dummy workerawaitaio.sleep(1/n)returnnasyncdefcb(res,err,ctx):# callbackiferr:# error handlingexc,tb=errasserttb# the only purpose of this is loggingreturnexcpool,n=ctx# context can be anything you likeawaitaio.sleep(1/ (n-1))returnres+ntodo=range(5)futures= []asyncwithAioPool(size=2)aspool:foriintodo:fut=pool.spawn_n(wrk(i),cb, (pool,i))futures.append(fut)results= []forfutinfutures:# there are helpers for result extraction. `flat` one will do# exactly what's written below#   from asyncio_pool import getres#   results.append(getres.flat(fut))try:results.append(fut.result())exceptExceptionase:results.append(e)# First error happens for n == 0 in wrk, exception of it is passed to# callback, callback returns it to us. Second one happens in callback itself# and is passed to us by pool.assertall(isinstance(e,ZeroDivisionError)foreinresults[:2])# All n's in `todo` are passed through `wrk` and `cb` (cb adds wrk result# and # number, passed inside context), except for n == 0 and n == 1.assertsum(results[2:])==2* (sum(todo)-0-1)asyncdefexec_usage(todo=range(1,11)):asyncwithAioPool(size=4)aspool:futures=pool.map_n(worker,todo)# While other workers are waiting or active, you can "synchronously"# execute one task. It does not interrupt  others, just waits for pool# space, then waits for task to finish and then returns it's result.important_res=awaitpool.exec(worker(2))assert2==important_res# You can continue working as usual:moar=awaitpool.spawn(worker(10))assertsum(todo)==sum(f.result()forfinfutures)asyncdefcancel_usage():asyncdefwrk(*arg,**kw):awaitaio.sleep(0.5)return1pool=AioPool(size=2)f_quick=pool.spawn_n(aio.sleep(0.1))f12=awaitpool.spawn(wrk()),pool.spawn_n(wrk())f35=pool.map_n(wrk,range(3))# At this point, if you cancel futures, returned by pool methods,# you just won't be able to retrieve spawned task results, task# themselves will continue working. Don't do this:#   f_quick.cancel()# use `pool.cancel` instead:# cancel someawaitaio.sleep(0.1)cancelled,results=awaitpool.cancel(f12[0],f35[2])# running and waitingassert2==cancelled# none of them had time to finishassert2==len(results)and \all(isinstance(res,aio.CancelledError)forresinresults)# cancel all othersawaitaio.sleep(0.1)# not interrupted and finished successfullyassertf_quick.done()andf_quick.result()isNonecancelled,results=awaitpool.cancel()# allassert3==cancelledassertlen(results)==3and \all(isinstance(res,aio.CancelledError)forresinresults)assertawaitpool.join()# joins successfullyasyncdefdetails(todo=range(1,11)):pool=AioPool(size=5)# This code:f1= []foriintodo:f1.append(pool.spawn_n(worker(i)))# is equivalent to one call of `map_n`:f2=pool.map_n(worker,todo)# Afterwards you can await for any given future:try:assert3==awaitf1[2]# result of spawn_n(worker(3))exceptExceptionase:# exception happened in worker (or CancelledError) will be re-raisedpass# Or use `asyncio.wait` to handle results in batches (see `iterwait` also):important_res=0more_important= [f1[1],f2[1],f2[2]]whilemore_important:done,more_important=awaitaio.wait(more_important,timeout=0.5)# handle result, note it will re-raise exceptionsimportant_res+=sum(f.result()forfindone)assertimportant_res==2+2+3# But you need to join, to allow all spawned workers to finish# (of course you can `asyncio.wait` all of the futures if you want to)awaitpool.join()assertall(f.done()forfinitertools.chain(f1,f2))# this is guaranteedassert2*sum(todo)==sum(f.result()forfinitertools.chain(f1,f2))