o di @sdZddlZddlZddlmZmZmZmZddlm Z m Z m Z ddl m ZmZddlmZmZmZmZddd eeed e ed eefd d Zdddeeeded eefddZddddeededed ee fddZ  ddeeededed eefddZdS)zt Friendlier version of asyncio standard library. Provisional library. Must be imported as `aioitertools.asyncio`. N)AsyncGenerator AsyncIterable AwaitableIterable)AnycastOptional)iter maybe_await) AnyIterable AsyncIteratorMaybeAwaitableT)timeoutawsrreturncCst}dd|D}d}|r|dkrt|}nd}|rk|r?|t}|dkr?|D]}t|tjr9|q, q,ttttt t tt t ftj ||tj dIdH\}}|D]}|IdHVq^|sdSdS)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).0arrH/home/ubuntu/.local/lib/python3.10/site-packages/aioitertools/asyncio.py (szas_completed..Nr)r return_when) settime isinstancerFuturecancel TimeoutErrorrtuplerrwaitFIRST_COMPLETED)rrdonepending remaining thresholdfutitemrrr as_completeds6        r+F)return_exceptions iterablesr,csttdttddffdd fdd|D}t|}z~zK|roz}|r3|Vn|Wn tjy@Ynwz }|VWn"tjylt|D] }|ra| |qVt dIdHYnw|s(Wn tj t fy|YnwW|D] }|s| q|D]}z|IdHWqtj yYqwdS|D] }|s| q|D]}z|IdHWqtj yYqww) 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 r rNc sz|2z3dHW}|IdHq6WdStjy+t|tr*|IdHtyE}z|IdHWYd}~dSd}~wwN)putrCancelledErrorrraclose Exception)r r*e) exc_queuequeuerrtaileros zas_generated..tailercsg|] }t|qSrr)rr )r6rr zsz as_generated..gMbP?)rQueuerrr get_nowait QueueEmptylistr%removesleepr0 GeneratorExitr )r-r,tasksr&excvaluetaskr)r4r5r6r as_generatedOsn    rCr,limitargsrFc si}i}dgt|}t}t}d} |t|kr`|dks%t||kr`|||vr5||||nt||} || ||| <|g|||<|d7}|t|kr`|dks%t||ks%|rz+tj|tjdIdH\}}|D]} |r| r| ||| <qr| ||| <qrWntj y|D]} | qtj |ddiIdHw|s|t|krnq| D]} tdt| D] } || d|| | <qq|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) NrTrDr )rr,)lenrappendrraddr#r$ exceptionresultr0r gathervaluesrange) r,rFrG input_mapposretr&r%next_argrBxlstirrrrMsR       )rMitrcs,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. cs g|z 3dHW}t|q6Sr.)r )rrVrrrr7szgather_iter..NrE)rMaiter)rWr,rFrrr gather_iters  rY)FrD)__doc__rrcollections.abcrrrrtypingrrrbuiltinsr rXr typesr r rrfloatr+boolrCintr;rMrYrrrrsZ  ?  S Q