U l/e @sdZddlZddlZddlmZmZmZmZmZm Z m Z m Z m Z m Z mZddlmZmZddlmZmZmZmZdde eee eeedd d Zd d e eeeeed ddZd ddeeeee edddZdeeeeee edddZdS)zt Friendlier version of asyncio standard library. Provisional library. Must be imported as `aioitertools.asyncio`. N) AnyAsyncGenerator AsyncIterable AwaitablecastDictIterableListOptionalSetTuple)iter maybe_await) AnyIterable AsyncIteratorMaybeAwaitableT)timeout)awsrreturncCst}dd|D}d}|r2|dkr2t|}nd}|r|rz|t}|dkrz|D]}t|tjrV|qVqVtttt t t t t t ftj ||tj dIdH\}}|D]}|IdHVqq6dS)a Run awaitables in `aws` concurrently, and yield results as they complete. Unlike `asyncio.as_completed`, this yields actual results, and does not require awaiting each item in the iterable. Cancels all remaining awaitables if a timeout is given and the timeout threshold is reached. Example:: async for value in as_completed(futures): ... # use value immediately cSsh|]}t|qSasyncio ensure_future).0arr8/tmp/pip-unpacked-wheel-n_uj7zou/aioitertools/asyncio.py 3szas_completed..Nr)r return_when)settime isinstancerZFuturecancel TimeoutErrorrr r rrwaitFIRST_COMPLETED)rrdonepending remaining thresholdZfutitemrrr as_completeds2       r,F)return_exceptions) iterablesr-rcsVttttddfdd fdd|D}t|}zz|rz}|rb|Vn|Wntj k r~YnXz}|VWqHtj k rt |D]}|r| |qt dIdHYqHXqHWntjt fk rYnXW5|D]}|s|q|D].}z|IdHWntjk rJYnXq XdS)ao Yield results from one or more async iterables, in the order they are produced. Like :func:`as_completed`, but for async iterators or generators instead of futures. Creates a separate task to drain each iterable, and a single queue for results. If ``return_exceptions`` is ``False``, then any exception will be raised, and pending iterables and tasks will be cancelled, and async generators will be closed. If ``return_exceptions`` is ``True``, any exceptions will be yielded as results, and execution will continue until all iterables have been fully consumed. Example:: async def generator(x): for i in range(x): yield i gen1 = generator(10) gen2 = generator(12) async for value in as_generated([gen1, gen2]): ... # intermixed values yielded from gen1 and gen2 N)rrc sz(|2z3dHW}|IdHq6Wn`tjk rXt|trR|IdHYn2tk r}z|IdHW5d}~XYnXdSN)putrCancelledErrorr"raclose Exception)rr+e) exc_queuequeuerrtailerzs zas_generated..tailercsg|]}t|qSrr)rr)r7rr sz as_generated..gMbP?)rQueuerrr r'r#r1 get_nowaitZ QueueEmptylistremovesleep GeneratorExit)r.r-Ztasksr(taskexcvaluer)r5r6r7r as_generatedZs@       rBr-limit)argsr-rErc si}i}dgt|}t}t}d}|t|kr|dksFt||kr|||krf||||n.t||} || ||| <|g|||<|d7}q&|r@zVtj|tjdIdH\}}|D]2} |r| r| ||| <q| ||| <qWnDtj k r>|D]} | qtj |ddiIdHYnX|s&|t|kr&qVq&| D]0} tdt| D]} || d|| | <qpq^|S)aR Like asyncio.gather but with a limit on concurrency. Note that all results are buffered. If gather is cancelled all tasks that were internally created and still pending will be cancelled as well. Example:: futures = [some_coro(i) for i in range(10)] results = await gather(*futures, limit=2) NrrCr )rr-T)lenr appendrraddr%r& exceptionresultr1r#gathervaluesrange) r-rErFZ input_mapposretr(r'next_argr?xlstirrrrLsF       rL)itrr-rErcs*tddt|2IdH||dIdHS)z Wrapper around gather to handle gathering an iterable instead of *args. Note that the iterable values don't have to be awaitable. csg|z3dHW}t|q6Sr/)r)rrTrrrr8szgather_iter..NrD)rLaiter)rUr-rErrr gather_iters rW)FrC)__doc__rr!typingrrrrrrrr r r r builtinsrrVrtypesrrrrfloatr,boolrBintrLrWrrrrsB4  ?  S Q