async_checkpointer() — langchain Function Reference

Architecture documentation for the async_checkpointer() function in conftest.py from the langchain codebase.

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ async_checkpointer() — langchain Function Reference

Dependency Diagram

graph TD
  22e0ab09_648a_a45c_1fc4_9cd73f6bfda7["async_checkpointer()"]
  81b9f9df_8601_314e_4fb6_20a115260a63["conftest.py"]
  22e0ab09_648a_a45c_1fc4_9cd73f6bfda7 -->|defined in| 81b9f9df_8601_314e_4fb6_20a115260a63
  style 22e0ab09_648a_a45c_1fc4_9cd73f6bfda7 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/tests/unit_tests/agents/conftest.py lines 173–194

async def async_checkpointer(
    request: pytest.FixtureRequest,
) -> AsyncIterator[BaseCheckpointSaver[str]]:
    checkpointer_name = request.param
    if checkpointer_name == "memory":
        with _checkpointer_memory() as checkpointer:
            yield checkpointer
    elif checkpointer_name == "sqlite_aio":
        async with _checkpointer_sqlite_aio() as checkpointer:
            yield checkpointer
    elif checkpointer_name == "postgres_aio":
        async with _checkpointer_postgres_aio() as checkpointer:
            yield checkpointer
    elif checkpointer_name == "postgres_aio_pipe":
        async with _checkpointer_postgres_aio_pipe() as checkpointer:
            yield checkpointer
    elif checkpointer_name == "postgres_aio_pool":
        async with _checkpointer_postgres_aio_pool() as checkpointer:
            yield checkpointer
    else:
        msg = f"Unknown checkpointer: {checkpointer_name}"
        raise NotImplementedError(msg)