Implementation:Langchain ai Langgraph SqliteSaver From Conn String

Attribute	Value
Page Type	Implementation (API Doc)
Library	langgraph (checkpoint-sqlite, checkpoint-postgres)
Workflow	Persistence_and_Memory_Setup
Principle	Langchain_ai_Langgraph_Checkpointer_Initialization
Implementation	SqliteSaver_From_Conn_String
Source	`libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py:L38-557`, `libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py:L32-103`

Overview

SqliteSaver.from_conn_string and PostgresSaver.from_conn_string are factory class methods that create fully configured checkpoint savers from database connection strings. Both return context managers that handle connection lifecycle automatically.

Description

These factory methods encapsulate the boilerplate of creating database connections with appropriate settings and wrapping them in checkpoint saver instances. They are the recommended way to initialize database-backed checkpointers for most use cases.

Usage

Import

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver

Code Reference

Source Location

Class	File	Lines
`SqliteSaver`	`libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py`	L38-557
`PostgresSaver`	`libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py`	L32-103

SqliteSaver Signatures

class SqliteSaver(BaseCheckpointSaver[str]):
    conn: sqlite3.Connection
    is_setup: bool

    def __init__(
        self,
        conn: sqlite3.Connection,
        *,
        serde: SerializerProtocol | None = None,
    ) -> None: ...

    @classmethod
    @contextmanager
    def from_conn_string(cls, conn_string: str) -> Iterator[SqliteSaver]:
        """Create a new SqliteSaver instance from a connection string.

        Args:
            conn_string: The SQLite connection string.

        Yields:
            SqliteSaver: A new SqliteSaver instance.
        """
        ...

    def setup(self) -> None:
        """Set up the checkpoint database.
        Creates necessary tables if they don't exist.
        Called automatically when needed.
        """
        ...

PostgresSaver Signatures

class PostgresSaver(BasePostgresSaver):
    lock: threading.Lock

    def __init__(
        self,
        conn: _internal.Conn,
        pipe: Pipeline | None = None,
        serde: SerializerProtocol | None = None,
    ) -> None: ...

    @classmethod
    @contextmanager
    def from_conn_string(
        cls, conn_string: str, *, pipeline: bool = False
    ) -> Iterator[PostgresSaver]:
        """Create a new PostgresSaver instance from a connection string.

        Args:
            conn_string: The Postgres connection info string.
            pipeline: whether to use Pipeline

        Returns:
            PostgresSaver: A new PostgresSaver instance.
        """
        ...

    def setup(self) -> None:
        """Set up the checkpoint database.
        Creates necessary tables and runs migrations.
        Must be called explicitly by the user.
        """
        ...

I/O Contract

SqliteSaver.from_conn_string

Parameter	Type	Description
`conn_string`	`str`	SQLite connection string. Use `":memory:"` for in-memory database, or a file path like `"checkpoints.sqlite"` for disk persistence.
Yields	Type	Description
	`SqliteSaver`	Configured checkpoint saver with an active SQLite connection.

PostgresSaver.from_conn_string

Parameter	Type	Description
`conn_string`	`str`	PostgreSQL connection string, e.g. `"postgresql://user:pass@localhost:5432/dbname"`.
`pipeline`	`bool`	Whether to use pipeline mode for batching SQL statements. Default: `False`.
Yields	Type	Description
	`PostgresSaver`	Configured checkpoint saver with an active PostgreSQL connection.

SQLite Table Schema (auto-created)

CREATE TABLE IF NOT EXISTS checkpoints (
    thread_id TEXT NOT NULL,
    checkpoint_ns TEXT NOT NULL DEFAULT '',
    checkpoint_id TEXT NOT NULL,
    parent_checkpoint_id TEXT,
    type TEXT,
    checkpoint BLOB,
    metadata BLOB,
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
);

CREATE TABLE IF NOT EXISTS writes (
    thread_id TEXT NOT NULL,
    checkpoint_ns TEXT NOT NULL DEFAULT '',
    checkpoint_id TEXT NOT NULL,
    task_id TEXT NOT NULL,
    idx INTEGER NOT NULL,
    channel TEXT NOT NULL,
    type TEXT,
    value BLOB,
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
);

Usage Examples

SQLite: In-Memory Database

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph

builder = StateGraph(int)
builder.add_node("add_one", lambda x: x + 1)
builder.set_entry_point("add_one")
builder.set_finish_point("add_one")

with SqliteSaver.from_conn_string(":memory:") as memory:
    graph = builder.compile(checkpointer=memory)
    config = {"configurable": {"thread_id": "1"}}
    result = graph.invoke(3, config)
    state = graph.get_state(config)

SQLite: Persistent File

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("checkpoints.sqlite") as memory:
    graph = builder.compile(checkpointer=memory)
    # Graph state persists across process restarts
    config = {"configurable": {"thread_id": "conversation-1"}}
    result = graph.invoke(inputs, config)

SQLite: Direct Connection

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
memory = SqliteSaver(conn)
graph = builder.compile(checkpointer=memory)

PostgreSQL: Basic Setup

from langgraph.checkpoint.postgres import PostgresSaver

conn_string = "postgresql://user:pass@localhost:5432/mydb"

with PostgresSaver.from_conn_string(conn_string) as checkpointer:
    checkpointer.setup()  # Run migrations (required on first use)
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "thread-1"}}
    result = graph.invoke(inputs, config)

PostgreSQL: With Pipeline Mode

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string(conn_string, pipeline=True) as checkpointer:
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)
    result = graph.invoke(inputs, config)

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment