The missing async toolbox - re-implements functions and classes of the Python standard library to make them compatible with async callables, iterables and context managers
84
Quality
Pending
Does it follow best practices?
Impact
84%
3.36xAverage score across 10 eval scenarios
Async versions of Python's built-in functions for iteration, aggregation, and data processing. These functions provide the fundamental async building blocks that mirror Python's standard builtins while working seamlessly with both sync and async iterables.
Functions for navigating and consuming async iterators.
async def anext(iterator: AsyncIterator[T]) -> T: ...
async def anext(iterator: AsyncIterator[T], default: T) -> T:
"""
Retrieve the next item from the async iterator.
Parameters:
- iterator: AsyncIterator[T] - The async iterator to get next item from
- default: T, optional - Value to return if iterator is exhausted
Returns:
T - The next item from the iterator
Raises:
StopAsyncIteration - If iterator is exhausted and no default provided
"""Usage example:
async def example():
async def async_gen():
yield 1
yield 2
it = aiter(async_gen())
first = await anext(it) # 1
second = await anext(it) # 2
third = await anext(it, "done") # "done" (default value)Create async iterators from various input types.
def iter(subject: AnyIterable[T]) -> AsyncIterator[T]: ...
def iter(subject: Callable[[], Awaitable[T | None]], sentinel: None) -> AsyncIterator[T]: ...
def iter(subject: Callable[[], Awaitable[T]], sentinel: T) -> AsyncIterator[T]:
"""
Create an async iterator from subject.
Parameters:
- subject: AnyIterable[T] or Callable - Iterable or callable to create iterator from
- sentinel: T, optional - Sentinel value for callable subjects
Returns:
AsyncIterator[T] - Async iterator yielding items from subject
"""Transform and enumerate items from async iterables.
def zip(*, strict: bool = False) -> AsyncIterator[Any]: ...
def zip(__it1: AnyIterable[T1], *, strict: bool = False) -> AsyncIterator[tuple[T1]]: ...
def zip(__it1: AnyIterable[T1], __it2: AnyIterable[T2], *, strict: bool = False) -> AsyncIterator[tuple[T1, T2]]: ...
def zip(__it1: AnyIterable[T1], __it2: AnyIterable[T2], __it3: AnyIterable[T3], *, strict: bool = False) -> AsyncIterator[tuple[T1, T2, T3]]: ...
def zip(__it1: AnyIterable[T1], __it2: AnyIterable[T2], __it3: AnyIterable[T3], __it4: AnyIterable[T4], *, strict: bool = False) -> AsyncIterator[tuple[T1, T2, T3, T4]]: ...
def zip(__it1: AnyIterable[T1], __it2: AnyIterable[T2], __it3: AnyIterable[T3], __it4: AnyIterable[T4], __it5: AnyIterable[T5], *, strict: bool = False) -> AsyncIterator[tuple[T1, T2, T3, T4, T5]]: ...
def zip(__it1: AnyIterable[Any], __it2: AnyIterable[Any], __it3: AnyIterable[Any], __it4: AnyIterable[Any], __it5: AnyIterable[Any], *iterables: AnyIterable[Any], strict: bool = False) -> AsyncIterator[tuple[Any, ...]]:
"""
Zip multiple iterables together.
Parameters:
- *iterables: AnyIterable - Variable number of iterables to zip
- strict: bool - If True, iterables must have same length
Returns:
AsyncIterator[tuple] - Iterator of tuples containing items from each iterable
"""
def map(function: Callable[[T1], Awaitable[R]], __it1: AnyIterable[T1], /) -> AsyncIterator[R]: ...
def map(function: Callable[[T1], R], __it1: AnyIterable[T1], /) -> AsyncIterator[R]: ...
def map(function: Callable[[T1, T2], Awaitable[R]], __it1: AnyIterable[T1], __it2: AnyIterable[T2], /) -> AsyncIterator[R]: ...
def map(function: Callable[[T1, T2], R], __it1: AnyIterable[T1], __it2: AnyIterable[T2], /) -> AsyncIterator[R]: ...
def map(function: Callable[[T1, T2, T3], Awaitable[R]], __it1: AnyIterable[T1], __it2: AnyIterable[T2], __it3: AnyIterable[T3], /) -> AsyncIterator[R]: ...
def map(function: Callable[[T1, T2, T3], R], __it1: AnyIterable[T1], __it2: AnyIterable[T2], __it3: AnyIterable[T3], /) -> AsyncIterator[R]: ...
def map(function: Callable[..., Awaitable[R]], __it1: AnyIterable[Any], __it2: AnyIterable[Any], __it3: AnyIterable[Any], __it4: AnyIterable[Any], __it5: AnyIterable[Any], /, *iterable: AnyIterable[Any]) -> AsyncIterator[R]: ...
def map(function: Callable[..., R], __it1: AnyIterable[Any], __it2: AnyIterable[Any], __it3: AnyIterable[Any], __it4: AnyIterable[Any], __it5: AnyIterable[Any], /, *iterable: AnyIterable[Any]) -> AsyncIterator[R]:
"""
Apply function to items from iterables.
Parameters:
- function: Callable - Function to apply (can be sync or async)
- *iterables: AnyIterable - Iterables to map over
Returns:
AsyncIterator[R] - Iterator of function results
"""
def filter(function: None, iterable: AnyIterable[T | None]) -> AsyncIterator[T]: ...
def filter(function: Callable[[T], TypeGuard[R]], iterable: AnyIterable[T]) -> AsyncIterator[R]: ...
def filter(function: Callable[[T], Any], iterable: AnyIterable[T]) -> AsyncIterator[T]:
"""
Filter items based on predicate function.
Parameters:
- function: Callable[[T], bool] or None - Predicate function (None filters falsy values)
- iterable: AnyIterable[T] - Iterable to filter
Returns:
AsyncIterator[T] - Iterator of items where predicate is True
"""
def enumerate(iterable: AnyIterable[T], start: int = 0) -> AsyncIterator[tuple[int, T]]:
"""
Return enumerated async iterator with indices.
Parameters:
- iterable: AnyIterable[T] - Iterable to enumerate
- start: int - Starting index value
Returns:
AsyncIterator[tuple[int, T]] - Iterator of (index, item) tuples
"""Aggregate values from async iterables.
async def all(iterable: AnyIterable[Any]) -> bool:
"""
Check if all elements are truthy.
Parameters:
- iterable: AnyIterable[Any] - Iterable to check
Returns:
bool - True if all elements are truthy, False otherwise
"""
async def any(iterable: AnyIterable[Any]) -> bool:
"""
Check if any element is truthy.
Parameters:
- iterable: AnyIterable[Any] - Iterable to check
Returns:
bool - True if any element is truthy, False otherwise
"""
async def max(iterable: AnyIterable[LT], *, key: None = None) -> LT: ...
async def max(iterable: AnyIterable[LT], *, key: None = None, default: T) -> LT | T: ...
async def max(iterable: AnyIterable[T1], *, key: Callable[[T1], LT]) -> T1: ...
async def max(iterable: AnyIterable[T1], *, key: Callable[[T1], LT], default: T2) -> T1 | T2:
"""
Find maximum value in iterable.
Parameters:
- iterable: AnyIterable[T] - Iterable to find max in
- key: Callable[[T], Any], optional - Key function for comparison
- default: T, optional - Value to return if iterable is empty
Returns:
T - Maximum value
Raises:
ValueError - If iterable is empty and no default provided
"""
async def min(iterable: AnyIterable[LT], *, key: None = None) -> LT: ...
async def min(iterable: AnyIterable[LT], *, key: None = None, default: T) -> LT | T: ...
async def min(iterable: AnyIterable[T1], *, key: Callable[[T1], LT]) -> T1: ...
async def min(iterable: AnyIterable[T1], *, key: Callable[[T1], LT], default: T2) -> T1 | T2:
"""
Find minimum value in iterable.
Parameters:
- iterable: AnyIterable[T] - Iterable to find min in
- key: Callable[[T], Any], optional - Key function for comparison
- default: T, optional - Value to return if iterable is empty
Returns:
T - Minimum value
Raises:
ValueError - If iterable is empty and no default provided
"""
async def sum(iterable: AnyIterable[int]) -> int: ...
async def sum(iterable: AnyIterable[float]) -> float: ...
async def sum(iterable: AnyIterable[ADD], start: ADD) -> ADD:
"""
Sum numeric values from iterable.
Parameters:
- iterable: AnyIterable[Numeric] - Iterable of numeric values
- start: Numeric - Starting value for summation
Returns:
Numeric - Sum of all values plus start
"""Convert async iterables to Python collection types.
async def list() -> list[Any]: ...
async def list(iterable: AnyIterable[T]) -> list[T]:
"""
Convert async iterable to list.
Parameters:
- iterable: AnyIterable[T] - Iterable to convert
Returns:
list[T] - List containing all items from iterable
"""
async def dict() -> dict[Any, Any]: ...
async def dict(iterable: AnyIterable[tuple[HK, T]]) -> dict[HK, T]: ...
async def dict(iterable: AnyIterable[tuple[str, T]] = (), **kwargs: T) -> dict[str, T]:
"""
Convert async iterable to dict.
Parameters:
- iterable: AnyIterable[tuple[K, V]] - Iterable of key-value pairs
- **kwargs: V - Additional keyword arguments as key-value pairs
Returns:
dict[K, V] - Dictionary from iterable and kwargs
"""
async def set() -> set[Any]: ...
async def set(iterable: AnyIterable[T] = ()) -> set[T]:
"""
Convert async iterable to set.
Parameters:
- iterable: AnyIterable[T] - Iterable to convert
Returns:
set[T] - Set containing unique items from iterable
"""
async def tuple() -> tuple[()]: ...
async def tuple(iterable: AnyIterable[T]) -> tuple[T, ...]:
"""
Convert async iterable to tuple.
Parameters:
- iterable: AnyIterable[T] - Iterable to convert
Returns:
tuple[T, ...] - Tuple containing all items from iterable
"""
async def sorted(iterable: AnyIterable[LT], *, key: None = None, reverse: bool = False) -> list[LT]: ...
async def sorted(iterable: AnyIterable[T], *, key: Callable[[T], LT], reverse: bool = False) -> list[T]:
"""
Return sorted list from iterable.
Parameters:
- iterable: AnyIterable[T] - Iterable to sort
- key: Callable[[T], Any], optional - Key function for sorting
- reverse: bool - If True, sort in descending order
Returns:
list[T] - Sorted list of items
"""async def aggregate_example():
async def numbers():
for i in range(10):
yield i
total = await sum(numbers()) # 45
all_positive = await all(map(lambda x: x >= 0, numbers())) # True
has_five = await any(map(lambda x: x == 5, numbers())) # Trueasync def collect_example():
async def pairs():
for i in range(5):
yield (f"key_{i}", i * 2)
items = await list(pairs()) # [('key_0', 0), ('key_1', 2), ...]
lookup = await dict(pairs()) # {'key_0': 0, 'key_1': 2, ...}
values = await set(map(lambda p: p[1], pairs())) # {0, 2, 4, 6, 8}Install with Tessl CLI
npx tessl i tessl/pypi-asyncstdlibevals
scenario-1
scenario-2
scenario-3
scenario-4
scenario-5
scenario-6
scenario-7
scenario-8
scenario-9
scenario-10