CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-asyncstdlib

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

3.36x

Quality

Pending

Does it follow best practices?

Impact

84%

3.36x

Average score across 10 eval scenarios

Overview
Eval results
Files

builtins.mddocs/

Built-in Functions

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.

Capabilities

Async Iterator Navigation

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)

Iterator Creation

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
    """

Iteration and Enumeration

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
    """

Aggregation Functions

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
    """

Collection Construction

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
    """

Usage Examples

Basic Aggregation

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()))  # True

Collection Building

async 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-asyncstdlib

docs

asynctools.md

builtins.md

contextlib.md

functools.md

heapq.md

index.md

itertools.md

tile.json