Checkpointer API

LangGraph checkpoint savers for CockroachDB.

CockroachDBSaver

Synchronous checkpointer for LangGraph workflows.

langchain_cockroachdb.checkpointer.saver.CockroachDBSaver

Bases: PsycopgOps, BaseCockroachDBSaver

Synchronous checkpointer that stores checkpoints in CockroachDB.

Uses raw psycopg3 connections (not SQLAlchemy) for compatibility with LangGraph's checkpointer interface. For production, pass a psycopg_pool.ConnectionPool: it replaces broken connections, while a single connection cannot recover from a dropped connection.

On a pool, operations run concurrently, each on its own pooled connection. On a single connection they run one at a time. A write with several statements runs them in one transaction, so it is saved completely or not at all, even when the caller stops partway through. On a single connection the caller also uses, an operation inside a transaction the caller opened joins that transaction, and its errors are raised at once, not retried, so the caller can retry the whole transaction.

Example

from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool
from langchain_cockroachdb import CockroachDBSaver, RetryPolicy

pool = ConnectionPool(
    "postgresql://root@localhost:26257/defaultdb?sslmode=disable",
    kwargs={"autocommit": True, "row_factory": dict_row},
    open=True,
)
checkpointer = CockroachDBSaver(pool, retry_policy=RetryPolicy(max_attempts=8))
checkpointer.setup()
graph = workflow.compile(checkpointer=checkpointer)

delete_thread(thread_id)

Delete all checkpoints and writes for a thread.

Parameters:

Name	Type	Description	Default
thread_id	str	The thread ID to delete.	required

disable_ttl()

Disable CockroachDB row-level TTL on checkpoint tables.

Safe to run from several processes at once. On CockroachDB v25.4 a TTL change fails while another one runs on the same table; that error is retried under the saver's retry policy, even if its retry_on would refuse it. max_attempts=1 turns this off.

enable_ttl(ttl_interval='7 days', cron='@daily')

Enable CockroachDB row-level TTL on checkpoint tables.

Rows older than ttl_interval will be automatically deleted by a background CockroachDB job on the specified cron schedule.

Safe to run from several processes at once. On CockroachDB v25.4 a TTL change fails while another one runs on the same table; that error is retried under the saver's retry policy, even if its retry_on would refuse it. max_attempts=1 turns this off.

Parameters:

Name	Type	Description	Default
ttl_interval	str	Interval after which rows expire (e.g., '7 days', '30 days').	'7 days'
cron	str	Cron schedule for the TTL deletion job (default: '@daily').	'@daily'

from_conn_string(conn_string, *, pipeline=False, serde=None, retry_policy=None, connect_timeout=10.0, prepare_threshold=5, connect_kwargs=None) classmethod

Create a saver on a single connection.

Parameters:

Name	Type	Description	Default
conn_string	str	CockroachDB connection string (cockroachdb://, postgresql:// and similar)	required
pipeline	bool	Whether to use Pipeline mode	False
serde	SerializerProtocol | None	Custom serializer protocol	None
retry_policy	RetryPolicy | None	How operations retry (default: RetryPolicy())	None
connect_timeout	float | None	Seconds to wait when connecting, or None (default: 10)	10.0
prepare_threshold	int | None	psycopg prepared statement threshold, None to disable (default: 5). Statements sent in a pipeline are never prepared	5
connect_kwargs	Mapping[str, Any] | None	Extra psycopg connection options, e.g. application_name	None

Yields:

Type	Description
CockroachDBSaver	CockroachDBSaver instance.

get_tuple(config)

Get a checkpoint tuple from the database.

Parameters:

Name	Type	Description	Default
config	RunnableConfig	Config identifying the checkpoint.	required

Returns:

Type	Description
CheckpointTuple | None	CheckpointTuple or None if not found.

list(config, *, filter=None, before=None, limit=None)

List checkpoints from the database.

Matching checkpoints are fetched inside one retried call and yielded afterwards, so no connection or lock is held while the caller iterates.

Parameters:

Name	Type	Description	Default
config	RunnableConfig | None	Config for filtering checkpoints.	required
filter	dict[str, Any] | None	Additional metadata filtering criteria.	None
before	RunnableConfig | None	Only return checkpoints before this checkpoint ID.	None
limit	int | None	Maximum number of checkpoints to return.	None

Yields:

Type	Description
CheckpointTuple	CheckpointTuple for each matching checkpoint.

put(config, checkpoint, metadata, new_versions)

Save a checkpoint to the database.

Parameters:

Name	Type	Description	Default
config	RunnableConfig	Config to associate with the checkpoint.	required
checkpoint	Checkpoint	The checkpoint to save.	required
metadata	CheckpointMetadata	Additional metadata.	required
new_versions	ChannelVersions	New channel versions as of this write.	required

Returns:

Type	Description
RunnableConfig	Updated configuration with checkpoint ID.

put_writes(config, writes, task_id, task_path='')

Store intermediate writes linked to a checkpoint.

Parameters:

Name	Type	Description	Default
config	RunnableConfig	Configuration of the related checkpoint.	required
writes	Sequence[tuple[str, Any]]	List of (channel, value) pairs to store.	required
task_id	str	Identifier for the task creating the writes.	required
task_path	str	Path of the task creating the writes.	''

setup()

Set up the checkpoint database.

Creates necessary tables and runs migrations. MUST be called the first time the checkpointer is used. Safe to run from several processes at once: every migration is idempotent and versions are recorded with ON CONFLICT DO NOTHING.

AsyncCockroachDBSaver

Asynchronous checkpointer for LangGraph workflows.

langchain_cockroachdb.checkpointer.async_saver.AsyncCockroachDBSaver

Bases: AsyncPsycopgOps, BaseCockroachDBSaver

Asynchronous checkpointer that stores checkpoints in CockroachDB.

Uses raw psycopg3 async connections (not SQLAlchemy) for compatibility with LangGraph's checkpointer interface. For production, pass a psycopg_pool.AsyncConnectionPool: it replaces broken connections, while a single connection cannot recover from a dropped connection.

On a pool, operations run concurrently, each on its own pooled connection. On a single connection they run one at a time. A write with several statements runs them in one transaction, so it is saved completely or not at all, even when the caller is cancelled partway through. On a single connection the caller also uses, an operation inside a transaction the caller opened joins that transaction, and its errors are raised at once, not retried, so the caller can retry the whole transaction.

The sync methods (get_tuple, put and the others) run the async ones on the event loop the saver was created on, so call them from other threads while that loop runs. From the loop itself they raise asyncio.InvalidStateError. When the loop is closed or stops, or in a forked child, they raise RuntimeError instead of waiting forever.

Example

from psycopg.rows import dict_row
from psycopg_pool import AsyncConnectionPool
from langchain_cockroachdb import AsyncCockroachDBSaver, RetryPolicy

async with AsyncConnectionPool(
    "postgresql://root@localhost:26257/defaultdb?sslmode=disable",
    kwargs={"autocommit": True, "row_factory": dict_row},
) as pool:
    checkpointer = AsyncCockroachDBSaver(pool, retry_policy=RetryPolicy(max_attempts=8))
    await checkpointer.setup()
    graph = workflow.compile(checkpointer=checkpointer)

adelete_thread(thread_id) async

Delete all checkpoints and writes for a thread.

Parameters:

Name	Type	Description	Default
thread_id	str	The thread ID to delete.	required

adisable_ttl() async

Disable CockroachDB row-level TTL on checkpoint tables.

Safe to run from several processes at once. On CockroachDB v25.4 a TTL change fails while another one runs on the same table; that error is retried under the saver's retry policy, even if its retry_on would refuse it. max_attempts=1 turns this off.

aenable_ttl(ttl_interval='7 days', cron='@daily') async

Enable CockroachDB row-level TTL on checkpoint tables.

Safe to run from several processes at once. On CockroachDB v25.4 a TTL change fails while another one runs on the same table; that error is retried under the saver's retry policy, even if its retry_on would refuse it. max_attempts=1 turns this off.

Parameters:

Name	Type	Description	Default
ttl_interval	str	Interval after which rows expire (e.g., '7 days', '30 days').	'7 days'
cron	str	Cron schedule for the TTL deletion job (default: '@daily').	'@daily'

aget_tuple(config) async

Get a checkpoint tuple from the database asynchronously.

Parameters:

Name	Type	Description	Default
config	RunnableConfig	Config identifying the checkpoint.	required

Returns:

Type	Description
CheckpointTuple | None	CheckpointTuple or None if not found.

alist(config, *, filter=None, before=None, limit=None) async

List checkpoints from the database asynchronously.

Matching checkpoints are fetched inside one retried call and yielded afterwards, so no connection or lock is held while the caller iterates.

Parameters:

Name	Type	Description	Default
config	RunnableConfig | None	Base configuration for filtering checkpoints.	required
filter	dict[str, Any] | None	Additional metadata filtering criteria.	None
before	RunnableConfig | None	Only return checkpoints before this checkpoint ID.	None
limit	int | None	Maximum number of checkpoints to return.	None

Yields:

Type	Description
AsyncIterator[CheckpointTuple]	CheckpointTuple for each matching checkpoint.

aput(config, checkpoint, metadata, new_versions) async

Save a checkpoint to the database asynchronously.

Parameters:

Name	Type	Description	Default
config	RunnableConfig	Config to associate with the checkpoint.	required
checkpoint	Checkpoint	The checkpoint to save.	required
metadata	CheckpointMetadata	Additional metadata.	required
new_versions	ChannelVersions	New channel versions as of this write.	required

Returns:

Type	Description
RunnableConfig	Updated configuration with checkpoint ID.

aput_writes(config, writes, task_id, task_path='') async

Store intermediate writes linked to a checkpoint asynchronously.

Parameters:

Name	Type	Description	Default
config	RunnableConfig	Configuration of the related checkpoint.	required
writes	Sequence[tuple[str, Any]]	List of (channel, value) pairs to store.	required
task_id	str	Identifier for the task creating the writes.	required
task_path	str	Path of the task creating the writes.	''

delete_thread(thread_id)

Delete thread (sync wrapper for adelete_thread).

from_conn_string(conn_string, *, pipeline=False, serde=None, retry_policy=None, connect_timeout=10.0, prepare_threshold=5, connect_kwargs=None) async classmethod

Create a saver on a single connection.

Parameters:

Name	Type	Description	Default
conn_string	str	CockroachDB connection string (cockroachdb://, postgresql:// and similar)	required
pipeline	bool	Whether to use AsyncPipeline mode	False
serde	SerializerProtocol | None	Custom serializer protocol	None
retry_policy	RetryPolicy | None	How operations retry (default: RetryPolicy())	None
connect_timeout	float | None	Seconds to wait when connecting, or None (default: 10)	10.0
prepare_threshold	int | None	psycopg prepared statement threshold, None to disable (default: 5). Statements sent in a pipeline are never prepared	5
connect_kwargs	Mapping[str, Any] | None	Extra psycopg connection options, e.g. application_name	None

Yields:

Type	Description
AsyncIterator[AsyncCockroachDBSaver]	AsyncCockroachDBSaver instance.

get_tuple(config)

Get a checkpoint tuple (sync wrapper for aget_tuple).

list(config, *, filter=None, before=None, limit=None)

List checkpoints (sync wrapper for alist).

put(config, checkpoint, metadata, new_versions)

Save a checkpoint (sync wrapper for aput).

put_writes(config, writes, task_id, task_path='')

Store intermediate writes (sync wrapper for aput_writes).

setup() async

Set up the checkpoint database asynchronously.

Creates necessary tables and runs migrations. MUST be called the first time the checkpointer is used. Safe to run from several processes at once: every migration is idempotent and versions are recorded with ON CONFLICT DO NOTHING.

Parameters

Parameter	Type	Default	Description
conn	Connection or Pool	Required	psycopg connection or pool
retry_policy	RetryPolicy	None	Retry policy for transient errors
pipe	bool	False	Use pipeline mode (single connection)

from_conn_string Options

Parameter	Type	Default	Description
conn_string	str	Required	Database connection string (cockroachdb:// or postgresql://)
retry_policy	RetryPolicy	None	Retry policy for transient errors
connect_timeout	float | None	10	Connection timeout in seconds
prepare_threshold	int	5	Prepared statement threshold
connect_kwargs	dict	{}	Additional connection options (e.g. {"application_name": "my-service"})
pipeline	bool	False	Use pipeline mode (single connection)

Key Methods

Shared (Sync & Async)

Method	Async Method	Description
setup()	setup()	Create tables and run migrations
put()	aput()	Save a checkpoint
get_tuple()	aget_tuple()	Retrieve a checkpoint
list()	alist()	List checkpoints with optional filters
put_writes()	aput_writes()	Store intermediate writes
delete_thread()	adelete_thread()	Delete all checkpoints for a thread
enable_ttl(ttl_interval, cron)	aenable_ttl(ttl_interval, cron)	Enable CockroachDB row-level TTL
disable_ttl()	adisable_ttl()	Disable CockroachDB row-level TTL

Factory Methods

Method	Description
from_conn_string(conn_string)	Create saver from a connection string (context manager, prepare_threshold=5)

Connection Types

Both savers accept these connection types:

Type	Description
psycopg.Connection / AsyncConnection	Single connection (simple scripts)
psycopg_pool.ConnectionPool / AsyncConnectionPool	Connection pool (production)

Examples

See Checkpointer Guide for comprehensive usage examples.