test_default_atransform_with_dicts() — langchain Function Reference

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

Function python LangChainCore Runnables calls 1

Entity Profile

LangChainCore→ Runnables→ test_default_atransform_with_dicts() — langchain Function Reference

Dependency Diagram

graph TD
  2c1ab2e6_eda2_c95d_aeb2_128b4062ea59["test_default_atransform_with_dicts()"]
  26df6ad8_0189_51d0_c3c1_6c3248893ff5["test_runnable.py"]
  2c1ab2e6_eda2_c95d_aeb2_128b4062ea59 -->|defined in| 26df6ad8_0189_51d0_c3c1_6c3248893ff5
  fb618d44_c03b_ea8b_385b_2278dfb173d4["invoke()"]
  2c1ab2e6_eda2_c95d_aeb2_128b4062ea59 -->|calls| fb618d44_c03b_ea8b_385b_2278dfb173d4
  style 2c1ab2e6_eda2_c95d_aeb2_128b4062ea59 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 5527–5556

async def test_default_atransform_with_dicts() -> None:
    """Test that default transform works with dicts."""

    class CustomRunnable(RunnableSerializable[Input, Output]):
        @override
        def invoke(
            self, input: Input, config: RunnableConfig | None = None, **kwargs: Any
        ) -> Output:
            return cast("Output", input)

    runnable = CustomRunnable[dict[str, str], dict[str, str]]()

    async def chunk_iterator() -> AsyncIterator[dict[str, str]]:
        yield {"foo": "a"}
        yield {"foo": "n"}

    chunks = [chunk async for chunk in runnable.atransform(chunk_iterator())]

    assert chunks == [{"foo": "n"}]

    # Test with addable dict
    async def chunk_iterator_with_addable() -> AsyncIterator[dict[str, str]]:
        yield AddableDict({"foo": "a"})
        yield AddableDict({"foo": "n"})

    chunks = [
        chunk async for chunk in runnable.atransform(chunk_iterator_with_addable())
    ]

    assert chunks == [{"foo": "an"}]