Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def map(fn: Callable[[T], R], itr: AnyIterable[T]) -> AsyncIterator[R]:
"""
Modify item of a mixed iterable using the given function or coroutine.
Example:
async for response in map(func, data):
...
"""
# todo: queue items eagerly
async for item in iter(itr):
yield await maybe_await(fn(item))
async for key, group in groupby(data, key=str.lower):
key # "a", "b", "c"
group # ["A", "a"], ["b"], ["c", "C", "c"]
"""
if key is None:
key = lambda x: x
grouping: List[T] = []
it = iter(itr)
item = await next(it)
grouping = [item]
j = await maybe_await(key(item))
async for item in it:
k = await maybe_await(key(item))
if k != j:
yield j, grouping
grouping = [item]
else:
grouping.append(item)
j = k
yield j, grouping
Yield values from a function using an iterable of iterables for arguments.
Each iterable contained within will be unpacked and consumed before
executing the function or coroutine.
Example:
data = [(1, 1), (1, 1, 1), (2, 2)]
async for value in starmap(operator.add, data):
... # 2, 3, 4
"""
async for itr in iter(iterable):
args = await list(itr)
yield await maybe_await(fn(*args))
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 values from the iterable until the predicate evaluates False.
Accepts both standard functions and coroutines for the predicate.
Example:
def pred(x):
return x < 4
async for value in takewhile(pred, range(8)):
... # 0, 1, 2, 3
"""
async for item in iter(iterable):
if await maybe_await(predicate(item)):
yield item
else:
break