aiter.py — langchain Source File

Source Code

"""Asynchronous iterator utilities.

Adapted from
https://github.com/maxfischer2781/asyncstdlib/blob/master/asyncstdlib/itertools.py
MIT License.
"""

from collections import deque
from collections.abc import (
    AsyncGenerator,
    AsyncIterable,
    AsyncIterator,
    Awaitable,
    Callable,
    Iterator,
)
from contextlib import AbstractAsyncContextManager
from types import TracebackType
from typing import (
    Any,
    Generic,
    TypeVar,
    cast,
    overload,
)

from typing_extensions import override

from langchain_core._api.deprecation import deprecated

T = TypeVar("T")

_no_default = object()


# https://github.com/python/cpython/blob/main/Lib/test/test_asyncgen.py#L54
@deprecated(since="1.1.2", removal="2.0.0")
def py_anext(
    iterator: AsyncIterator[T], default: T | Any = _no_default
) -> Awaitable[T | Any | None]:
    """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.

    Args:
        iterator: The async iterator to advance.
        default: The value to return if the iterator is exhausted.

            If not provided, a `StopAsyncIteration` exception is raised.

    Returns:
        The next value from the iterator, or the default value if the iterator is
            exhausted.

    Raises:
        TypeError: If the iterator is not an async iterator.
// ... (288 more lines)