Source code for langchain.utils.aiter
"""
Adapted from
https://github.com/maxfischer2781/asyncstdlib/blob/master/asyncstdlib/itertools.py
MIT License
"""
from collections import deque
from typing import (
    Any,
    AsyncContextManager,
    AsyncGenerator,
    AsyncIterator,
    Awaitable,
    Callable,
    Deque,
    Generic,
    Iterator,
    List,
    Optional,
    Tuple,
    TypeVar,
    Union,
    cast,
    overload,
)
T = TypeVar("T")
_no_default = object()
# https://github.com/python/cpython/blob/main/Lib/test/test_asyncgen.py#L54
# before 3.10, the builtin anext() was not available
[docs]def py_anext(
    iterator: AsyncIterator[T], default: Union[T, Any] = _no_default
) -> Awaitable[Union[T, None, Any]]:
    """Pure-Python implementation of anext() for testing purposes.
    Closely matches the builtin anext() C implementation.
    Can be used to compare the built-in implementation of the inner
    coroutines machinery to C-implementation of __anext__() and send()
    or throw() on the returned generator.
    """
    try:
        __anext__ = cast(
            Callable[[AsyncIterator[T]], Awaitable[T]], type(iterator).__anext__
        )
    except AttributeError:
        raise TypeError(f"{iterator!r} is not an async iterator")
    if default is _no_default:
        return __anext__(iterator)
    async def anext_impl() -> Union[T, Any]:
        try:
            # The C code is way more low-level than this, as it implements