test_async_retry_batch_preserves_order() — langchain Function Reference

Architecture documentation for the test_async_retry_batch_preserves_order() function in test_runnable.py from the langchain codebase.

Function python LangChainCore Runnables calls 1

Entity Profile

LangChainCore→ Runnables→ test_async_retry_batch_preserves_order() — langchain Function Reference

Dependency Diagram

graph TD
  a881619b_3d89_0144_9cc5_f86b0287ead9["test_async_retry_batch_preserves_order()"]
  26df6ad8_0189_51d0_c3c1_6c3248893ff5["test_runnable.py"]
  a881619b_3d89_0144_9cc5_f86b0287ead9 -->|defined in| 26df6ad8_0189_51d0_c3c1_6c3248893ff5
  8652094c_ec57_c551_fc44_9566d00cf872["abatch()"]
  a881619b_3d89_0144_9cc5_f86b0287ead9 -->|calls| 8652094c_ec57_c551_fc44_9566d00cf872
  style a881619b_3d89_0144_9cc5_f86b0287ead9 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 3960–3979

async def test_async_retry_batch_preserves_order() -> None:
    """Async variant of order preservation regression test."""
    first_fail: set[int] = {1}

    def sometimes_fail(x: int) -> int:  # pragma: no cover - trivial
        if x in first_fail:
            first_fail.remove(x)
            msg = "fail once"
            raise ValueError(msg)
        return x

    runnable = RunnableLambda(sometimes_fail)

    results = await runnable.with_retry(
        stop_after_attempt=2,
        wait_exponential_jitter=False,
        retry_if_exception_type=(ValueError,),
    ).abatch([0, 1, 2])

    assert results == [0, 1, 2]