Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Example:
data = [1, 2, 3, 4]
async def mul(a, b):
return a * b
async for total in accumulate(data, func=mul):
... # 1, 2, 6, 24
"""
itr = iter(itr)
try:
total: T = await next(itr)
except AnyStop:
return
yield total
async for item in itr:
total = await maybe_await(func(total, item))
yield total
"""
Yield a tuple of items from mixed iterables until the shortest is consumed.
Example:
async for a, b, c in zip(i, j, k):
...
"""
its: List[AsyncIterator[Any]] = [iter(itr) for itr in itrs]
while True:
try:
values = await asyncio.gather(*[it.__anext__() for it in its])
yield values
except AnyStop:
break
async for a, b in zip_longest(a, b, fillvalue=-1):
a # 0, 1, 2, -1, -1
b # 0, 1, 2, 3, 4
"""
its: List[AsyncIterator[Any]] = [iter(itr) for itr in itrs]
itr_count = len(its)
finished = 0
while True:
values = await asyncio.gather(
*[it.__anext__() for it in its], return_exceptions=True
)
for idx, value in builtins.enumerate(values):
if isinstance(value, AnyStop):
finished += 1
values[idx] = fillvalue
its[idx] = repeat(fillvalue)
elif isinstance(value, BaseException):
raise value
if finished >= itr_count:
break
yield tuple(values)