test_abatch() — langchain Function Reference

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

Function python CoreAbstractions Serialization calls 1

Entity Profile

CoreAbstractions→ Serialization→ test_abatch() — langchain Function Reference

Dependency Diagram

graph TD
  2d240e6c_0dba_793d_c16e_cecb30f225e7["test_abatch()"]
  d62d478e_a330_c328_7da4_58e88d780807["test_fallbacks.py"]
  2d240e6c_0dba_793d_c16e_cecb30f225e7 -->|defined in| d62d478e_a330_c328_7da4_58e88d780807
  b4df7c29_3d15_7d1f_4ec5_5bcda5d0c06a["_assert_potential_error()"]
  2d240e6c_0dba_793d_c16e_cecb30f225e7 -->|calls| b4df7c29_3d15_7d1f_4ec5_5bcda5d0c06a
  style 2d240e6c_0dba_793d_c16e_cecb30f225e7 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_fallbacks.py lines 210–257

async def test_abatch() -> None:
    runnable = RunnableLambda(_runnable)
    with pytest.raises(ValueError, match="missing exception"):
        await runnable.abatch([{"text": "foo"}, {"text": "bar"}, {"text": "baz"}])
    actual = await runnable.abatch(
        [{"text": "foo"}, {"text": "bar"}, {"text": "baz"}], return_exceptions=True
    )
    expected = ["first", ValueError(), ValueError()]
    _assert_potential_error(actual, expected)

    runnable_with_single = runnable.with_fallbacks(
        [runnable], exception_key="exception"
    )
    with pytest.raises(RuntimeError):
        await runnable_with_single.abatch(
            [
                {"text": "foo"},
                {"text": "bar"},
                {"text": "baz"},
            ]
        )
    actual = await runnable_with_single.abatch(
        [{"text": "foo"}, {"text": "bar"}, {"text": "baz"}], return_exceptions=True
    )
    expected = ["first", "second", RuntimeError()]
    _assert_potential_error(actual, expected)

    runnable_with_double = runnable.with_fallbacks(
        [runnable, runnable], exception_key="exception"
    )
    actual = await runnable_with_double.abatch(
        [{"text": "foo"}, {"text": "bar"}, {"text": "baz"}], return_exceptions=True
    )

    expected = ["first", "second", "third"]
    _assert_potential_error(actual, expected)

    runnable_with_double = runnable.with_fallbacks(
        [runnable, runnable],
        exception_key="exception",
        exceptions_to_handle=(ValueError,),
    )
    actual = await runnable_with_double.abatch(
        [{"text": "foo"}, {"text": "bar"}, {"text": "baz"}], return_exceptions=True
    )

    expected = ["first", "second", RuntimeError()]
    _assert_potential_error(actual, expected)