sync_checkpointer() — langchain Function Reference

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

Function python LangChainCore ApiManagement

Entity Profile

LangChainCore→ ApiManagement→ sync_checkpointer() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/langchain_v1/tests/unit_tests/agents/conftest.py lines 146–167

def sync_checkpointer(
    request: pytest.FixtureRequest,
) -> Iterator[BaseCheckpointSaver[str]]:
    checkpointer_name = request.param
    if checkpointer_name == "memory":
        with _checkpointer_memory() as checkpointer:
            yield checkpointer
    elif checkpointer_name == "sqlite":
        with _checkpointer_sqlite() as checkpointer:
            yield checkpointer
    elif checkpointer_name == "postgres":
        with _checkpointer_postgres() as checkpointer:
            yield checkpointer
    elif checkpointer_name == "postgres_pipe":
        with _checkpointer_postgres_pipe() as checkpointer:
            yield checkpointer
    elif checkpointer_name == "postgres_pool":
        with _checkpointer_postgres_pool() as checkpointer:
            yield checkpointer
    else:
        msg = f"Unknown checkpointer: {checkpointer_name}"
        raise NotImplementedError(msg)