test_abatch() — langchain Function Reference

Architecture documentation for the test_abatch() function in test_indexing.py from the langchain codebase.

Function python LangChainCore ApiManagement calls 1

Entity Profile

LangChainCore→ ApiManagement→ test_abatch() — langchain Function Reference

Dependency Diagram

graph TD
  10935246_0fd9_e2cf_9188_e739f3c6e6d8["test_abatch()"]
  a9fb4c74_0865_0941_ade3_563a79762cee["test_indexing.py"]
  10935246_0fd9_e2cf_9188_e739f3c6e6d8 -->|defined in| a9fb4c74_0865_0941_ade3_563a79762cee
  899b9749_c7a0_c31e_0e55_9f07b6d8b44f["_to_async_iter()"]
  10935246_0fd9_e2cf_9188_e739f3c6e6d8 -->|calls| 899b9749_c7a0_c31e_0e55_9f07b6d8b44f
  style 10935246_0fd9_e2cf_9188_e739f3c6e6d8 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/indexing/test_indexing.py lines 2417–2433

async def test_abatch() -> None:
    """Test the abatch function."""
    batches = _abatch(5, _to_async_iter(range(12)))
    assert isinstance(batches, AsyncIterator)
    assert [batch async for batch in batches] == [
        [0, 1, 2, 3, 4],
        [5, 6, 7, 8, 9],
        [10, 11],
    ]

    batches = _abatch(1, _to_async_iter(range(3)))
    assert isinstance(batches, AsyncIterator)
    assert [batch async for batch in batches] == [[0], [1], [2]]

    batches = _abatch(2, _to_async_iter(range(5)))
    assert isinstance(batches, AsyncIterator)
    assert [batch async for batch in batches] == [[0, 1], [2, 3], [4]]