async_store() — langchain Function Reference

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

Function python LangChainCore MessageInterface

Entity Profile

LangChainCore→ MessageInterface→ async_store() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/langchain_v1/tests/unit_tests/agents/conftest.py lines 122–140

async def async_store(request: pytest.FixtureRequest) -> AsyncIterator[BaseStore | None]:
    store_name = request.param
    if store_name is None:
        yield None
    elif store_name == "in_memory":
        with _store_memory() as store:
            yield store
    elif store_name == "postgres_aio":
        async with _store_postgres_aio() as store:
            yield store
    elif store_name == "postgres_aio_pipe":
        async with _store_postgres_aio_pipe() as store:
            yield store
    elif store_name == "postgres_aio_pool":
        async with _store_postgres_aio_pool() as store:
            yield store
    else:
        msg = f"Unknown store {store_name}"
        raise NotImplementedError(msg)