LangGraph Checkpoint SingleStore

Implementation of LangGraph CheckpointSaver that uses SingleStore.

Dependencies

By default langgraph-checkpoint-singlestore installs singlestoredb package. You can install it with:

pip install singlestoredb

or

uv add singlestoredb

Usage

Important

When using SingleStore checkpointers for the first time, make sure to call .setup() method on them to create required tables. See example below.

Important

When manually creating SingleStore connections and passing them to SingleStoreSaver or AsyncSingleStoreSaver, make sure to include autocommit=True and results_type="dict". See example below.

Why these parameters are required:

• autocommit=True: Required for the .setup() method to properly commit the checkpoint tables to the database. Without this, table creation may not be persisted.
• results_type="dict": Required because the SingleStoreSaver implementation accesses database rows using dictionary-style syntax (e.g., row["column_name"]). The default tuple results only support index-based access (e.g., row[0]), which will cause TypeError exceptions when the checkpointer tries to access columns by name.

Example of incorrect usage:

# ❌ This will fail with TypeError during checkpointer operations
with singlestoredb.connect(DB_URI) as conn:  # Missing autocommit=True and results_type="dict"
    checkpointer = SingleStoreSaver(conn)
    checkpointer.setup()  # May not persist tables properly
    # Any operation that reads from database will fail with:
    # TypeError: tuple indices must be integers or slices, not str

from langgraph.checkpoint.singlestore import SingleStoreSaver

write_config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
read_config = {"configurable": {"thread_id": "1"}}

DB_URI = "admin:password@svc-host:port/database_name"
with SingleStoreSaver.from_conn_string(DB_URI) as checkpointer:
    # call .setup() the first time you're using the checkpointer
    checkpointer.setup()
    checkpoint = {
        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {
            "my_key": "meow",
            "node": "node"
        },
        "channel_versions": {
            "__start__": 2,
            "my_key": 3,
            "start:node": 3,
            "node": 3
        },
        "versions_seen": {
            "__input__": {},
            "__start__": {
            "__start__": 1
            },
            "node": {
            "start:node": 2
            }
        },
    }

    # store checkpoint
    checkpointer.put(write_config, checkpoint, {}, {})

    # load checkpoint
    checkpointer.get(read_config)

    # list checkpoints
    list(checkpointer.list(read_config))

Async

from langgraph.checkpoint.singlestore.aio import AsyncSingleStoreSaver

async with AsyncSingleStoreSaver.from_conn_string(DB_URI) as checkpointer:
    checkpoint = {
        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {
            "my_key": "meow",
            "node": "node"
        },
        "channel_versions": {
            "__start__": 2,
            "my_key": 3,
            "start:node": 3,
            "node": 3
        },
        "versions_seen": {
            "__input__": {},
            "__start__": {
            "__start__": 1
            },
            "node": {
            "start:node": 2
            }
        },
    }

    # store checkpoint
    await checkpointer.aput(write_config, checkpoint, {}, {})

    # load checkpoint
    await checkpointer.aget(read_config)

    # list checkpoints
    [c async for c in checkpointer.alist(read_config)]