mode.utils.aiter

Async iterator utilities: aiter, anext, etc.

Python gained aiter() and anext() as builtins in 3.10, long after this module was written. The versions here are kept because they are not the same functions:

• aiter also accepts a synchronous iterable, wrapping it so that it can be driven with async for. The builtin raises TypeError for anything that does not implement __aiter__.
• anext takes its default as *default rather than as a single positional argument.

Both names shadow the builtins for the rest of this module, so aiter below always means the dispatcher defined here.

AsyncIterWrapper

Bases: AsyncIterator[T]

Wrap regular Iterator into an AsyncIterator.

Source code in mode/utils/aiter.py

class AsyncIterWrapper(AsyncIterator[T]):
    """Wrap regular Iterator into an AsyncIterator."""

    def __init__(self, it: Iterator[T]) -> None:
        self._it: Iterator[T] = it

    def __aiter__(self) -> AsyncIterator[T]:
        return self

    async def __anext__(self) -> T:
        try:
            return next(self._it)
        except StopIteration as exc:
            raise StopAsyncIteration() from exc

    def __repr__(self) -> str:
        return f"<{type(self).__name__}: {self._it}>"

arange

Bases: AsyncIterable[int]

Async generator that counts like range.

Source code in mode/utils/aiter.py

class arange(AsyncIterable[int]):
    """Async generator that counts like `range`."""

    def __init__(
        self, *slice_args: Optional[int], **slice_kwargs: Optional[int]
    ) -> None:
        s = slice(*slice_args, **slice_kwargs)
        if s.stop is None:
            raise TypeError("arange() requires a stop argument")
        self.start: int = s.start or 0
        self.stop: int = s.stop
        self.step: int = s.step or 1
        self._range = range(self.start, self.stop, self.step)

    def count(self, n: int) -> int:
        return self._range.count(n)

    def index(self, n: int) -> int:
        return self._range.index(n)

    def __contains__(self, n: int) -> bool:
        return n in self._range

    def __aiter__(self) -> AsyncIterator[int]:
        return _ARangeIterator(self, iter(self._range))

aenumerate(it, start=0) async

async for version of enumerate.

Source code in mode/utils/aiter.py

async def aenumerate(
    it: AsyncIterable[T], start: int = 0
) -> AsyncIterator[tuple[int, T]]:
    """``async for`` version of ``enumerate``."""
    i = start
    async for item in it:
        yield i, item
        i += 1

aiter(it)

Create an async iterator from an async or synchronous iterable.

Unlike the aiter builtin added in Python 3.10, a synchronous iterable is accepted as well: it is wrapped in AsyncIterWrapper so that it can be consumed with async for.

>>> [x async for x in aiter([1, 2, 3])]
[1, 2, 3]

Raises:

Type	Description
TypeError	if the argument is neither an AsyncIterable nor an Iterable.

Notes

If the object is already an iterator, the iterator should return self when __aiter__ is called.

Source code in mode/utils/aiter.py

@singledispatch
def aiter(it: Any) -> AsyncIterator[object]:
    """Create an async iterator from an async *or* synchronous iterable.

    Unlike the `aiter` builtin added in Python 3.10, a synchronous
    iterable is accepted as well: it is wrapped in `AsyncIterWrapper` so
    that it can be consumed with `async for`.

    ```sh
    >>> [x async for x in aiter([1, 2, 3])]
    [1, 2, 3]
    ```

    Raises:
        TypeError: if the argument is neither an `AsyncIterable` nor an
            `Iterable`.

    Notes:
        If the object is already an iterator, the iterator
        should return self when ``__aiter__`` is called.
    """
    raise TypeError(f"{it!r} object is not an iterable")

alist(ait) async

Convert async generator to list.

Source code in mode/utils/aiter.py

async def alist(ait: AsyncIterator[T]) -> list[T]:
    """Convert async generator to list."""
    return [x async for x in ait]

anext(it, *default) async

Get next value from async iterator, or default if empty.

Differs from the anext builtin added in Python 3.10: the default is taken as *default, so passing no default and passing one are both handled by this single signature.

Raises:

Type	Description
	exc:StopAsyncIteration: if default is not defined and the async iterator is fully consumed.

Source code in mode/utils/aiter.py

async def anext(it: AsyncIterator[T], *default: Optional[T]) -> T:
    """Get next value from async iterator, or `default` if empty.

    Differs from the `anext` builtin added in Python 3.10: the default is
    taken as `*default`, so passing no default and passing one are both
    handled by this single signature.

    Raises:
        :exc:`StopAsyncIteration`: if default is not defined and
            the async iterator is fully consumed.
    """
    if default:
        try:
            return await it.__anext__()
        except StopAsyncIteration:
            return cast(T, default[0])
    return await it.__anext__()

aslice(ait, *slice_args) async

Extract slice from async generator.

Source code in mode/utils/aiter.py

async def aslice(ait: AsyncIterator[T], *slice_args: int) -> AsyncIterator[T]:
    """Extract slice from async generator."""
    s = slice(*slice_args)
    start = s.start or 0
    stop = s.stop or sys.maxsize
    step = s.step or 1
    it = iter(range(start, stop, step))
    try:
        nexti = next(it)
        async for i, item in aenumerate(ait):
            if i == nexti:
                yield item
                nexti = next(it)
    except StopIteration:
        return

chunks(it, n) async

Split an async iterator into chunks with n elements each.

Example:

# n == 2
>>> x = chunks(arange(10), 2)
>>> [item async for item in x]
[[0, 1], [2, 3], [4, 5], [6, 7], [8, 9], [10]]

# n == 3
>>> x = chunks(arange(10)), 3)
>>> [item async for item in x]
[[0, 1, 2], [3, 4, 5], [6, 7, 8], [9, 10]]

Source code in mode/utils/aiter.py

async def chunks(it: AsyncIterable[T], n: int) -> AsyncIterable[list[T]]:
    """Split an async iterator into chunks with `n` elements each.

    Example:

    ```sh
    # n == 2
    >>> x = chunks(arange(10), 2)
    >>> [item async for item in x]
    [[0, 1], [2, 3], [4, 5], [6, 7], [8, 9], [10]]

    # n == 3
    >>> x = chunks(arange(10)), 3)
    >>> [item async for item in x]
    [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9, 10]]
    ```
    """
    # `aiter` is a singledispatch function, so its return type cannot be
    # tied to the type of its argument.
    ait = cast(AsyncIterator[T], aiter(it))
    async for item in ait:
        yield [item] + [x async for x in aslice(ait, n - 1)]