Checkpointers

CheckpointMetadata

Bases: TypedDict

Metadata associated with a checkpoint.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

class CheckpointMetadata(TypedDict, total=False):
    """Metadata associated with a checkpoint."""

    source: Literal["input", "loop", "update", "fork"]
    """The source of the checkpoint.

    - "input": The checkpoint was created from an input to invoke/stream/batch.
    - "loop": The checkpoint was created from inside the pregel loop.
    - "update": The checkpoint was created from a manual state update.
    - "fork": The checkpoint was created as a copy of another checkpoint.
    """
    step: int
    """The step number of the checkpoint.

    -1 for the first "input" checkpoint.
    0 for the first "loop" checkpoint.
    ... for the nth checkpoint afterwards.
    """
    writes: dict[str, Any]
    """The writes that were made between the previous checkpoint and this one.

    Mapping from node name to writes emitted by that node.
    """
    parents: dict[str, str]
    """The IDs of the parent checkpoints.

    Mapping from checkpoint namespace to checkpoint ID.
    """

source: Literal['input', 'loop', 'update', 'fork'] instance-attribute

The source of the checkpoint.

• "input": The checkpoint was created from an input to invoke/stream/batch.
• "loop": The checkpoint was created from inside the pregel loop.
• "update": The checkpoint was created from a manual state update.
• "fork": The checkpoint was created as a copy of another checkpoint.

step: int instance-attribute

The step number of the checkpoint.

-1 for the first "input" checkpoint. 0 for the first "loop" checkpoint. ... for the nth checkpoint afterwards.

writes: dict[str, Any] instance-attribute

The writes that were made between the previous checkpoint and this one.

Mapping from node name to writes emitted by that node.

parents: dict[str, str] instance-attribute

The IDs of the parent checkpoints.

Mapping from checkpoint namespace to checkpoint ID.

Checkpoint

Bases: TypedDict

State snapshot at a given point in time.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

class Checkpoint(TypedDict):
    """State snapshot at a given point in time."""

    v: int
    """The version of the checkpoint format. Currently 1."""
    id: str
    """The ID of the checkpoint. This is both unique and monotonically
    increasing, so can be used for sorting checkpoints from first to last."""
    ts: str
    """The timestamp of the checkpoint in ISO 8601 format."""
    channel_values: dict[str, Any]
    """The values of the channels at the time of the checkpoint.
    Mapping from channel name to deserialized channel snapshot value.
    """
    channel_versions: ChannelVersions
    """The versions of the channels at the time of the checkpoint.
    The keys are channel names and the values are monotonically increasing
    version strings for each channel.
    """
    versions_seen: dict[str, ChannelVersions]
    """Map from node ID to map from channel name to version seen.
    This keeps track of the versions of the channels that each node has seen.
    Used to determine which nodes to execute next.
    """
    pending_sends: List[SendProtocol]
    """List of inputs pushed to nodes but not yet processed.
    Cleared by the next checkpoint."""

v: int instance-attribute

The version of the checkpoint format. Currently 1.

id: str instance-attribute

The ID of the checkpoint. This is both unique and monotonically increasing, so can be used for sorting checkpoints from first to last.

ts: str instance-attribute

The timestamp of the checkpoint in ISO 8601 format.

channel_values: dict[str, Any] instance-attribute

The values of the channels at the time of the checkpoint. Mapping from channel name to deserialized channel snapshot value.

channel_versions: ChannelVersions instance-attribute

The versions of the channels at the time of the checkpoint. The keys are channel names and the values are monotonically increasing version strings for each channel.

versions_seen: dict[str, ChannelVersions] instance-attribute

Map from node ID to map from channel name to version seen. This keeps track of the versions of the channels that each node has seen. Used to determine which nodes to execute next.

pending_sends: List[SendProtocol] instance-attribute

List of inputs pushed to nodes but not yet processed. Cleared by the next checkpoint.

BaseCheckpointSaver

Bases: Generic[V]

Base class for creating a graph checkpointer.

Checkpointers allow LangGraph agents to persist their state within and across multiple interactions.

Attributes:

• serde (SerializerProtocol) –

  Serializer for encoding/decoding checkpoints.

Note

When creating a custom checkpoint saver, consider implementing async versions to avoid blocking the main thread.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

class BaseCheckpointSaver(Generic[V]):
    """Base class for creating a graph checkpointer.

    Checkpointers allow LangGraph agents to persist their state
    within and across multiple interactions.

    Attributes:
        serde (SerializerProtocol): Serializer for encoding/decoding checkpoints.

    Note:
        When creating a custom checkpoint saver, consider implementing async
        versions to avoid blocking the main thread.
    """

    serde: SerializerProtocol = JsonPlusSerializer()

    def __init__(
        self,
        *,
        serde: Optional[SerializerProtocol] = None,
    ) -> None:
        self.serde = maybe_add_typed_methods(serde or self.serde)

    @property
    def config_specs(self) -> list[ConfigurableFieldSpec]:
        """Define the configuration options for the checkpoint saver.

        Returns:
            list[ConfigurableFieldSpec]: List of configuration field specs.
        """
        return [CheckpointThreadId, CheckpointNS, CheckpointId]

    def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
        """Fetch a checkpoint using the given configuration.

        Args:
            config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

        Returns:
            Optional[Checkpoint]: The requested checkpoint, or None if not found.
        """
        if value := self.get_tuple(config):
            return value.checkpoint

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Fetch a checkpoint tuple using the given configuration.

        Args:
            config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

        Returns:
            Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError

    def list(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
        """List checkpoints that match the given criteria.

        Args:
            config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria.
            before (Optional[RunnableConfig]): List checkpoints created before this configuration.
            limit (Optional[int]): Maximum number of checkpoints to return.

        Returns:
            Iterator[CheckpointTuple]: Iterator of matching checkpoint tuples.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Store a checkpoint with its configuration and metadata.

        Args:
            config (RunnableConfig): Configuration for the checkpoint.
            checkpoint (Checkpoint): The checkpoint to store.
            metadata (CheckpointMetadata): Additional metadata for the checkpoint.
            new_versions (ChannelVersions): New channel versions as of this write.

        Returns:
            RunnableConfig: Updated configuration after storing the checkpoint.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError

    def put_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[Tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Store intermediate writes linked to a checkpoint.

        Args:
            config (RunnableConfig): Configuration of the related checkpoint.
            writes (List[Tuple[str, Any]]): List of writes to store.
            task_id (str): Identifier for the task creating the writes.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError

    async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
        """Asynchronously fetch a checkpoint using the given configuration.

        Args:
            config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

        Returns:
            Optional[Checkpoint]: The requested checkpoint, or None if not found.
        """
        if value := await self.aget_tuple(config):
            return value.checkpoint

    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Asynchronously fetch a checkpoint tuple using the given configuration.

        Args:
            config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

        Returns:
            Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError

    async def alist(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncIterator[CheckpointTuple]:
        """Asynchronously list checkpoints that match the given criteria.

        Args:
            config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
            before (Optional[RunnableConfig]): List checkpoints created before this configuration.
            limit (Optional[int]): Maximum number of checkpoints to return.

        Returns:
            AsyncIterator[CheckpointTuple]: Async iterator of matching checkpoint tuples.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError
        yield

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Asynchronously store a checkpoint with its configuration and metadata.

        Args:
            config (RunnableConfig): Configuration for the checkpoint.
            checkpoint (Checkpoint): The checkpoint to store.
            metadata (CheckpointMetadata): Additional metadata for the checkpoint.
            new_versions (ChannelVersions): New channel versions as of this write.

        Returns:
            RunnableConfig: Updated configuration after storing the checkpoint.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError

    async def aput_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[Tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Asynchronously store intermediate writes linked to a checkpoint.

        Args:
            config (RunnableConfig): Configuration of the related checkpoint.
            writes (List[Tuple[str, Any]]): List of writes to store.
            task_id (str): Identifier for the task creating the writes.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError

    def get_next_version(self, current: Optional[V], channel: ChannelProtocol) -> V:
        """Generate the next version ID for a channel.

        Default is to use integer versions, incrementing by 1. If you override, you can use str/int/float versions,
        as long as they are monotonically increasing.

        Args:
            current (Optional[V]): The current version identifier (int, float, or str).
            channel (BaseChannel): The channel being versioned.

        Returns:
            V: The next version identifier, which must be increasing.
        """
        if isinstance(current, str):
            raise NotImplementedError
        elif current is None:
            return 1
        else:
            return current + 1

config_specs: list[ConfigurableFieldSpec] property

Define the configuration options for the checkpoint saver.

Returns:

• list[ConfigurableFieldSpec] –

  list[ConfigurableFieldSpec]: List of configuration field specs.

get(config: RunnableConfig) -> Optional[Checkpoint]

Fetch a checkpoint using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[Checkpoint] –

  Optional[Checkpoint]: The requested checkpoint, or None if not found.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := self.get_tuple(config):
        return value.checkpoint

get_tuple(config: RunnableConfig) -> Optional[CheckpointTuple]

Fetch a checkpoint tuple using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[CheckpointTuple] –

  Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

Raises:

• NotImplementedError –

  Implement this method in your custom checkpoint saver.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Fetch a checkpoint tuple using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

list(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> Iterator[CheckpointTuple]

List checkpoints that match the given criteria.

Parameters:

• config (Optional[RunnableConfig]) –

  Base configuration for filtering checkpoints.

• filter (Optional[Dict[str, Any]], default: None ) –

  Additional filtering criteria.

• before (Optional[RunnableConfig], default: None ) –

  List checkpoints created before this configuration.

• limit (Optional[int], default: None ) –

  Maximum number of checkpoints to return.

Returns:

• Iterator[CheckpointTuple] –

  Iterator[CheckpointTuple]: Iterator of matching checkpoint tuples.

Raises:

• NotImplementedError –

  Implement this method in your custom checkpoint saver.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

def list(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
    """List checkpoints that match the given criteria.

    Args:
        config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
        filter (Optional[Dict[str, Any]]): Additional filtering criteria.
        before (Optional[RunnableConfig]): List checkpoints created before this configuration.
        limit (Optional[int]): Maximum number of checkpoints to return.

    Returns:
        Iterator[CheckpointTuple]: Iterator of matching checkpoint tuples.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

put(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig

Store a checkpoint with its configuration and metadata.

Parameters:

• config (RunnableConfig) –

  Configuration for the checkpoint.

• checkpoint (Checkpoint) –

  The checkpoint to store.

• metadata (CheckpointMetadata) –

  Additional metadata for the checkpoint.

• new_versions (ChannelVersions) –

  New channel versions as of this write.

Returns:

• RunnableConfig ( RunnableConfig ) –

  Updated configuration after storing the checkpoint.

Raises:

• NotImplementedError –

  Implement this method in your custom checkpoint saver.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Store a checkpoint with its configuration and metadata.

    Args:
        config (RunnableConfig): Configuration for the checkpoint.
        checkpoint (Checkpoint): The checkpoint to store.
        metadata (CheckpointMetadata): Additional metadata for the checkpoint.
        new_versions (ChannelVersions): New channel versions as of this write.

    Returns:
        RunnableConfig: Updated configuration after storing the checkpoint.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

put_writes(config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str) -> None

Store intermediate writes linked to a checkpoint.

Parameters:

• config (RunnableConfig) –

  Configuration of the related checkpoint.

• writes (List[Tuple[str, Any]]) –

  List of writes to store.

• task_id (str) –

  Identifier for the task creating the writes.

Raises:

• NotImplementedError –

  Implement this method in your custom checkpoint saver.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

def put_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
) -> None:
    """Store intermediate writes linked to a checkpoint.

    Args:
        config (RunnableConfig): Configuration of the related checkpoint.
        writes (List[Tuple[str, Any]]): List of writes to store.
        task_id (str): Identifier for the task creating the writes.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

aget(config: RunnableConfig) -> Optional[Checkpoint] async

Asynchronously fetch a checkpoint using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[Checkpoint] –

  Optional[Checkpoint]: The requested checkpoint, or None if not found.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Asynchronously fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := await self.aget_tuple(config):
        return value.checkpoint

aget_tuple(config: RunnableConfig) -> Optional[CheckpointTuple] async

Asynchronously fetch a checkpoint tuple using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[CheckpointTuple] –

  Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

Raises:

• NotImplementedError –

  Implement this method in your custom checkpoint saver.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Asynchronously fetch a checkpoint tuple using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

alist(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> AsyncIterator[CheckpointTuple] async

Asynchronously list checkpoints that match the given criteria.

Parameters:

• config (Optional[RunnableConfig]) –

  Base configuration for filtering checkpoints.

• filter (Optional[Dict[str, Any]], default: None ) –

  Additional filtering criteria for metadata.

• before (Optional[RunnableConfig], default: None ) –

  List checkpoints created before this configuration.

• limit (Optional[int], default: None ) –

  Maximum number of checkpoints to return.

Returns:

• AsyncIterator[CheckpointTuple] –

  AsyncIterator[CheckpointTuple]: Async iterator of matching checkpoint tuples.

Raises:

• NotImplementedError –

  Implement this method in your custom checkpoint saver.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def alist(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> AsyncIterator[CheckpointTuple]:
    """Asynchronously list checkpoints that match the given criteria.

    Args:
        config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
        filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
        before (Optional[RunnableConfig]): List checkpoints created before this configuration.
        limit (Optional[int]): Maximum number of checkpoints to return.

    Returns:
        AsyncIterator[CheckpointTuple]: Async iterator of matching checkpoint tuples.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError
    yield

aput(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig async

Asynchronously store a checkpoint with its configuration and metadata.

Parameters:

• config (RunnableConfig) –

  Configuration for the checkpoint.

• checkpoint (Checkpoint) –

  The checkpoint to store.

• metadata (CheckpointMetadata) –

  Additional metadata for the checkpoint.

• new_versions (ChannelVersions) –

  New channel versions as of this write.

Returns:

• RunnableConfig ( RunnableConfig ) –

  Updated configuration after storing the checkpoint.

Raises:

• NotImplementedError –

  Implement this method in your custom checkpoint saver.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def aput(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Asynchronously store a checkpoint with its configuration and metadata.

    Args:
        config (RunnableConfig): Configuration for the checkpoint.
        checkpoint (Checkpoint): The checkpoint to store.
        metadata (CheckpointMetadata): Additional metadata for the checkpoint.
        new_versions (ChannelVersions): New channel versions as of this write.

    Returns:
        RunnableConfig: Updated configuration after storing the checkpoint.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

aput_writes(config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str) -> None async

Asynchronously store intermediate writes linked to a checkpoint.

Parameters:

• config (RunnableConfig) –

  Configuration of the related checkpoint.

• writes (List[Tuple[str, Any]]) –

  List of writes to store.

• task_id (str) –

  Identifier for the task creating the writes.

Raises:

• NotImplementedError –

  Implement this method in your custom checkpoint saver.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def aput_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
) -> None:
    """Asynchronously store intermediate writes linked to a checkpoint.

    Args:
        config (RunnableConfig): Configuration of the related checkpoint.
        writes (List[Tuple[str, Any]]): List of writes to store.
        task_id (str): Identifier for the task creating the writes.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

get_next_version(current: Optional[V], channel: ChannelProtocol) -> V

Generate the next version ID for a channel.

Default is to use integer versions, incrementing by 1. If you override, you can use str/int/float versions, as long as they are monotonically increasing.

Parameters:

• current (Optional[V]) –

  The current version identifier (int, float, or str).

• channel (BaseChannel) –

  The channel being versioned.

Returns:

• V ( V ) –

  The next version identifier, which must be increasing.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

def get_next_version(self, current: Optional[V], channel: ChannelProtocol) -> V:
    """Generate the next version ID for a channel.

    Default is to use integer versions, incrementing by 1. If you override, you can use str/int/float versions,
    as long as they are monotonically increasing.

    Args:
        current (Optional[V]): The current version identifier (int, float, or str).
        channel (BaseChannel): The channel being versioned.

    Returns:
        V: The next version identifier, which must be increasing.
    """
    if isinstance(current, str):
        raise NotImplementedError
    elif current is None:
        return 1
    else:
        return current + 1

create_checkpoint(checkpoint: Checkpoint, channels: Optional[Mapping[str, ChannelProtocol]], step: int, *, id: Optional[str] = None) -> Checkpoint

Create a checkpoint for the given channels.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

def create_checkpoint(
    checkpoint: Checkpoint,
    channels: Optional[Mapping[str, ChannelProtocol]],
    step: int,
    *,
    id: Optional[str] = None,
) -> Checkpoint:
    """Create a checkpoint for the given channels."""
    ts = datetime.now(timezone.utc).isoformat()
    if channels is None:
        values = checkpoint["channel_values"]
    else:
        values = {}
        for k, v in channels.items():
            if k not in checkpoint["channel_versions"]:
                continue
            try:
                values[k] = v.checkpoint()
            except EmptyChannelError:
                pass
    return Checkpoint(
        v=1,
        ts=ts,
        id=id or str(uuid6(clock_seq=step)),
        channel_values=values,
        channel_versions=checkpoint["channel_versions"],
        versions_seen=checkpoint["versions_seen"],
        pending_sends=checkpoint.get("pending_sends", []),
    )

SerializerProtocol

Bases: Protocol

Protocol for serialization and deserialization of objects.

• dumps: Serialize an object to bytes.
• dumps_typed: Serialize an object to a tuple (type, bytes).
• loads: Deserialize an object from bytes.
• loads_typed: Deserialize an object from a tuple (type, bytes).

Valid implementations include the pickle, json and orjson modules.

Source code in libs/checkpoint/langgraph/checkpoint/serde/base.py

class SerializerProtocol(Protocol):
    """Protocol for serialization and deserialization of objects.

    - `dumps`: Serialize an object to bytes.
    - `dumps_typed`: Serialize an object to a tuple (type, bytes).
    - `loads`: Deserialize an object from bytes.
    - `loads_typed`: Deserialize an object from a tuple (type, bytes).

    Valid implementations include the `pickle`, `json` and `orjson` modules.
    """

    def dumps(self, obj: Any) -> bytes: ...

    def dumps_typed(self, obj: Any) -> tuple[str, bytes]: ...

    def loads(self, data: bytes) -> Any: ...

    def loads_typed(self, data: tuple[str, bytes]) -> Any: ...

JsonPlusSerializer

Bases: SerializerProtocol

Source code in libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py

class JsonPlusSerializer(SerializerProtocol):
    def _encode_constructor_args(
        self,
        constructor: Union[Callable, type[Any]],
        *,
        method: Union[None, str, Sequence[Union[None, str]]] = None,
        args: Optional[Sequence[Any]] = None,
        kwargs: Optional[dict[str, Any]] = None,
    ) -> dict[str, Any]:
        out = {
            "lc": 2,
            "type": "constructor",
            "id": (*constructor.__module__.split("."), constructor.__name__),
        }
        if method is not None:
            out["method"] = method
        if args is not None:
            out["args"] = args
        if kwargs is not None:
            out["kwargs"] = kwargs
        return out

    def _default(self, obj: Any) -> Union[str, dict[str, Any]]:
        if isinstance(obj, Serializable):
            return cast(dict[str, Any], obj.to_json())
        elif hasattr(obj, "model_dump") and callable(obj.model_dump):
            return self._encode_constructor_args(
                obj.__class__, method=(None, "model_construct"), kwargs=obj.model_dump()
            )
        elif hasattr(obj, "dict") and callable(obj.dict):
            return self._encode_constructor_args(
                obj.__class__, method=(None, "construct"), kwargs=obj.dict()
            )
        elif hasattr(obj, "_asdict") and callable(obj._asdict):
            return self._encode_constructor_args(obj.__class__, kwargs=obj._asdict())
        elif isinstance(obj, pathlib.Path):
            return self._encode_constructor_args(pathlib.Path, args=obj.parts)
        elif isinstance(obj, re.Pattern):
            return self._encode_constructor_args(
                re.compile, args=(obj.pattern, obj.flags)
            )
        elif isinstance(obj, UUID):
            return self._encode_constructor_args(UUID, args=(obj.hex,))
        elif isinstance(obj, decimal.Decimal):
            return self._encode_constructor_args(decimal.Decimal, args=(str(obj),))
        elif isinstance(obj, (set, frozenset, deque)):
            return self._encode_constructor_args(type(obj), args=(tuple(obj),))
        elif isinstance(obj, (IPv4Address, IPv4Interface, IPv4Network)):
            return self._encode_constructor_args(obj.__class__, args=(str(obj),))
        elif isinstance(obj, (IPv6Address, IPv6Interface, IPv6Network)):
            return self._encode_constructor_args(obj.__class__, args=(str(obj),))

        elif isinstance(obj, datetime):
            return self._encode_constructor_args(
                datetime, method="fromisoformat", args=(obj.isoformat(),)
            )
        elif isinstance(obj, timezone):
            return self._encode_constructor_args(
                timezone,
                args=obj.__getinitargs__(),  # type: ignore[attr-defined]
            )
        elif isinstance(obj, ZoneInfo):
            return self._encode_constructor_args(ZoneInfo, args=(obj.key,))
        elif isinstance(obj, timedelta):
            return self._encode_constructor_args(
                timedelta, args=(obj.days, obj.seconds, obj.microseconds)
            )
        elif isinstance(obj, date):
            return self._encode_constructor_args(
                date, args=(obj.year, obj.month, obj.day)
            )
        elif isinstance(obj, time):
            return self._encode_constructor_args(
                time,
                args=(obj.hour, obj.minute, obj.second, obj.microsecond, obj.tzinfo),
                kwargs={"fold": obj.fold},
            )
        elif dataclasses.is_dataclass(obj):
            return self._encode_constructor_args(
                obj.__class__,
                kwargs={
                    field.name: getattr(obj, field.name)
                    for field in dataclasses.fields(obj)
                },
            )
        elif isinstance(obj, Enum):
            return self._encode_constructor_args(obj.__class__, args=(obj.value,))
        elif isinstance(obj, SendProtocol):
            return self._encode_constructor_args(
                obj.__class__, kwargs={"node": obj.node, "arg": obj.arg}
            )
        elif isinstance(obj, (bytes, bytearray)):
            return self._encode_constructor_args(
                obj.__class__, method="fromhex", args=(obj.hex(),)
            )
        elif isinstance(obj, BaseException):
            return repr(obj)
        else:
            raise TypeError(
                f"Object of type {obj.__class__.__name__} is not JSON serializable"
            )

    def _reviver(self, value: dict[str, Any]) -> Any:
        if (
            value.get("lc", None) == 2
            and value.get("type", None) == "constructor"
            and value.get("id", None) is not None
        ):
            try:
                # Get module and class name
                [*module, name] = value["id"]
                # Import module
                mod = importlib.import_module(".".join(module))
                # Import class
                cls = getattr(mod, name)
                # Instantiate class
                method = value.get("method")
                if isinstance(method, str):
                    methods = [getattr(cls, method)]
                elif isinstance(method, list):
                    methods = [
                        cls if method is None else getattr(cls, method)
                        for method in method
                    ]
                else:
                    methods = [cls]
                args = value.get("args")
                kwargs = value.get("kwargs")
                for method in methods:
                    try:
                        if isclass(method) and issubclass(method, BaseException):
                            return None
                        if args and kwargs:
                            return method(*args, **kwargs)
                        elif args:
                            return method(*args)
                        elif kwargs:
                            return method(**kwargs)
                        else:
                            return method()
                    except Exception:
                        continue
            except Exception:
                return None

        return LC_REVIVER(value)

    def dumps(self, obj: Any) -> bytes:
        return json.dumps(obj, default=self._default, ensure_ascii=False).encode(
            "utf-8", "ignore"
        )

    def dumps_typed(self, obj: Any) -> tuple[str, bytes]:
        if isinstance(obj, bytes):
            return "bytes", obj
        elif isinstance(obj, bytearray):
            return "bytearray", obj
        else:
            try:
                return "msgpack", _msgpack_enc(obj)
            except UnicodeEncodeError:
                return "json", self.dumps(obj)

    def loads(self, data: bytes) -> Any:
        return json.loads(data, object_hook=self._reviver)

    def loads_typed(self, data: tuple[str, bytes]) -> Any:
        type_, data_ = data
        if type_ == "bytes":
            return data_
        elif type_ == "bytearray":
            return bytearray(data_)
        elif type_ == "json":
            return self.loads(data_)
        elif type_ == "msgpack":
            return msgpack.unpackb(
                data_, ext_hook=_msgpack_ext_hook, strict_map_key=False
            )
        else:
            raise NotImplementedError(f"Unknown serialization type: {type_}")

MemorySaver

Bases: BaseCheckpointSaver[str], AbstractContextManager, AbstractAsyncContextManager

An in-memory checkpoint saver.

This checkpoint saver stores checkpoints in memory using a defaultdict.

Note

Only use MemorySaver for debugging or testing purposes. For production use cases we recommend installing langgraph-checkpoint-postgres and using PostgresSaver / AsyncPostgresSaver.

Parameters:

• serde (Optional[SerializerProtocol], default: None ) –

  The serializer to use for serializing and deserializing checkpoints. Defaults to None.

Examples:

    import asyncio

    from langgraph.checkpoint.memory import MemorySaver
    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")

    memory = MemorySaver()
    graph = builder.compile(checkpointer=memory)
    coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
    asyncio.run(coro)  # Output: 2

Source code in libs/checkpoint/langgraph/checkpoint/memory/__init__.py

class MemorySaver(
    BaseCheckpointSaver[str], AbstractContextManager, AbstractAsyncContextManager
):
    """An in-memory checkpoint saver.

    This checkpoint saver stores checkpoints in memory using a defaultdict.

    Note:
        Only use `MemorySaver` for debugging or testing purposes.
        For production use cases we recommend installing [langgraph-checkpoint-postgres](https://pypi.org/project/langgraph-checkpoint-postgres/) and using `PostgresSaver` / `AsyncPostgresSaver`.

    Args:
        serde (Optional[SerializerProtocol]): The serializer to use for serializing and deserializing checkpoints. Defaults to None.

    Examples:

            import asyncio

            from langgraph.checkpoint.memory import MemorySaver
            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")

            memory = MemorySaver()
            graph = builder.compile(checkpointer=memory)
            coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
            asyncio.run(coro)  # Output: 2
    """

    # thread ID ->  checkpoint NS -> checkpoint ID -> checkpoint mapping
    storage: defaultdict[
        str,
        dict[
            str, dict[str, tuple[tuple[str, bytes], tuple[str, bytes], Optional[str]]]
        ],
    ]
    writes: defaultdict[
        tuple[str, str, str], dict[tuple[str, int], tuple[str, str, tuple[str, bytes]]]
    ]

    def __init__(
        self,
        *,
        serde: Optional[SerializerProtocol] = None,
        factory: Type[defaultdict] = defaultdict,
    ) -> None:
        super().__init__(serde=serde)
        self.storage = factory(lambda: defaultdict(dict))
        self.writes = factory(dict)
        self.stack = ExitStack()
        if factory is not defaultdict:
            self.stack.enter_context(self.storage)  # type: ignore[arg-type]
            self.stack.enter_context(self.writes)  # type: ignore[arg-type]

    def __enter__(self) -> "MemorySaver":
        return self.stack.__enter__()

    def __exit__(
        self,
        exc_type: Optional[type[BaseException]],
        exc_value: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> Optional[bool]:
        return self.stack.__exit__(exc_type, exc_value, traceback)

    async def __aenter__(self) -> "MemorySaver":
        return self.stack.__enter__()

    async def __aexit__(
        self,
        __exc_type: Optional[type[BaseException]],
        __exc_value: Optional[BaseException],
        __traceback: Optional[TracebackType],
    ) -> Optional[bool]:
        return self.stack.__exit__(__exc_type, __exc_value, __traceback)

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the in-memory storage.

        This method retrieves a checkpoint tuple from the in-memory storage based on the
        provided config. If the config contains a "checkpoint_id" key, the checkpoint with
        the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
        if checkpoint_id := get_checkpoint_id(config):
            if saved := self.storage[thread_id][checkpoint_ns].get(checkpoint_id):
                checkpoint, metadata, parent_checkpoint_id = saved
                writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
                if parent_checkpoint_id:
                    sends = [
                        w[2]
                        for w in self.writes[
                            (thread_id, checkpoint_ns, parent_checkpoint_id)
                        ].values()
                        if w[1] == TASKS
                    ]
                else:
                    sends = []
                return CheckpointTuple(
                    config=config,
                    checkpoint={
                        **self.serde.loads_typed(checkpoint),
                        "pending_sends": [self.serde.loads_typed(s) for s in sends],
                    },
                    metadata=self.serde.loads_typed(metadata),
                    pending_writes=[
                        (id, c, self.serde.loads_typed(v)) for id, c, v in writes
                    ],
                    parent_config={
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": parent_checkpoint_id,
                        }
                    }
                    if parent_checkpoint_id
                    else None,
                )
        else:
            if checkpoints := self.storage[thread_id][checkpoint_ns]:
                checkpoint_id = max(checkpoints.keys())
                checkpoint, metadata, parent_checkpoint_id = checkpoints[checkpoint_id]
                writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
                if parent_checkpoint_id:
                    sends = [
                        w[2]
                        for w in self.writes[
                            (thread_id, checkpoint_ns, parent_checkpoint_id)
                        ].values()
                        if w[1] == TASKS
                    ]
                else:
                    sends = []
                return CheckpointTuple(
                    config={
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": checkpoint_id,
                        }
                    },
                    checkpoint={
                        **self.serde.loads_typed(checkpoint),
                        "pending_sends": [self.serde.loads_typed(s) for s in sends],
                    },
                    metadata=self.serde.loads_typed(metadata),
                    pending_writes=[
                        (id, c, self.serde.loads_typed(v)) for id, c, v in writes
                    ],
                    parent_config={
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": parent_checkpoint_id,
                        }
                    }
                    if parent_checkpoint_id
                    else None,
                )

    def list(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
        """List checkpoints from the in-memory storage.

        This method retrieves a list of checkpoint tuples from the in-memory storage based
        on the provided criteria.

        Args:
            config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
            before (Optional[RunnableConfig]): List checkpoints created before this configuration.
            limit (Optional[int]): Maximum number of checkpoints to return.

        Yields:
            Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.
        """
        thread_ids = (config["configurable"]["thread_id"],) if config else self.storage
        config_checkpoint_ns = (
            config["configurable"].get("checkpoint_ns") if config else None
        )
        config_checkpoint_id = get_checkpoint_id(config) if config else None
        for thread_id in thread_ids:
            for checkpoint_ns in self.storage[thread_id].keys():
                if (
                    config_checkpoint_ns is not None
                    and checkpoint_ns != config_checkpoint_ns
                ):
                    continue

                for checkpoint_id, (
                    checkpoint,
                    metadata_b,
                    parent_checkpoint_id,
                ) in sorted(
                    self.storage[thread_id][checkpoint_ns].items(),
                    key=lambda x: x[0],
                    reverse=True,
                ):
                    # filter by checkpoint ID from config
                    if config_checkpoint_id and checkpoint_id != config_checkpoint_id:
                        continue

                    # filter by checkpoint ID from `before` config
                    if (
                        before
                        and (before_checkpoint_id := get_checkpoint_id(before))
                        and checkpoint_id >= before_checkpoint_id
                    ):
                        continue

                    # filter by metadata
                    metadata = self.serde.loads_typed(metadata_b)
                    if filter and not all(
                        query_value == metadata.get(query_key)
                        for query_key, query_value in filter.items()
                    ):
                        continue

                    # limit search results
                    if limit is not None and limit <= 0:
                        break
                    elif limit is not None:
                        limit -= 1

                    writes = self.writes[
                        (thread_id, checkpoint_ns, checkpoint_id)
                    ].values()

                    if parent_checkpoint_id:
                        sends = [
                            w[2]
                            for w in self.writes[
                                (thread_id, checkpoint_ns, parent_checkpoint_id)
                            ].values()
                            if w[1] == TASKS
                        ]
                    else:
                        sends = []

                    yield CheckpointTuple(
                        config={
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": checkpoint_id,
                            }
                        },
                        checkpoint={
                            **self.serde.loads_typed(checkpoint),
                            "pending_sends": [self.serde.loads_typed(s) for s in sends],
                        },
                        metadata=metadata,
                        parent_config={
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": parent_checkpoint_id,
                            }
                        }
                        if parent_checkpoint_id
                        else None,
                        pending_writes=[
                            (id, c, self.serde.loads_typed(v)) for id, c, v in writes
                        ],
                    )

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Save a checkpoint to the in-memory storage.

        This method saves a checkpoint to the in-memory storage. The checkpoint is associated
        with the provided config.

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.
            metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
            new_versions (dict): New versions as of this write

        Returns:
            RunnableConfig: The updated config containing the saved checkpoint's timestamp.
        """
        c = checkpoint.copy()
        c.pop("pending_sends")  # type: ignore[misc]
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"]["checkpoint_ns"]
        self.storage[thread_id][checkpoint_ns].update(
            {
                checkpoint["id"]: (
                    self.serde.dumps_typed(c),
                    self.serde.dumps_typed(metadata),
                    config["configurable"].get("checkpoint_id"),  # parent
                )
            }
        )
        return {
            "configurable": {
                "thread_id": thread_id,
                "checkpoint_ns": checkpoint_ns,
                "checkpoint_id": checkpoint["id"],
            }
        }

    def put_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[Tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Save a list of writes to the in-memory storage.

        This method saves a list of writes to the in-memory storage. The writes are associated
        with the provided config.

        Args:
            config (RunnableConfig): The config to associate with the writes.
            writes (list[tuple[str, Any]]): The writes to save.
            task_id (str): Identifier for the task creating the writes.

        Returns:
            RunnableConfig: The updated config containing the saved writes' timestamp.
        """
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
        checkpoint_id = config["configurable"]["checkpoint_id"]
        outer_key = (thread_id, checkpoint_ns, checkpoint_id)
        outer_writes_ = self.writes.get(outer_key)
        for idx, (c, v) in enumerate(writes):
            inner_key = (task_id, WRITES_IDX_MAP.get(c, idx))
            if inner_key[1] >= 0 and outer_writes_ and inner_key in outer_writes_:
                continue

            self.writes[outer_key][inner_key] = (task_id, c, self.serde.dumps_typed(v))

    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Asynchronous version of get_tuple.

        This method is an asynchronous wrapper around get_tuple that runs the synchronous
        method in a separate thread using asyncio.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        return await asyncio.get_running_loop().run_in_executor(
            None, self.get_tuple, config
        )

    async def alist(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncIterator[CheckpointTuple]:
        """Asynchronous version of list.

        This method is an asynchronous wrapper around list that runs the synchronous
        method in a separate thread using asyncio.

        Args:
            config (RunnableConfig): The config to use for listing the checkpoints.

        Yields:
            AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples.
        """
        loop = asyncio.get_running_loop()
        iter = await loop.run_in_executor(
            None,
            partial(
                self.list,
                before=before,
                limit=limit,
                filter=filter,
            ),
            config,
        )
        while True:
            # handling StopIteration exception inside coroutine won't work
            # as expected, so using next() with default value to break the loop
            if item := await loop.run_in_executor(None, next, iter, None):
                yield item
            else:
                break

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Asynchronous version of put.

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.
            metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
            new_versions (dict): New versions as of this write

        Returns:
            RunnableConfig: The updated config containing the saved checkpoint's timestamp.
        """
        return await asyncio.get_running_loop().run_in_executor(
            None, self.put, config, checkpoint, metadata, new_versions
        )

    async def aput_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[Tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Asynchronous version of put_writes.

        This method is an asynchronous wrapper around put_writes that runs the synchronous
        method in a separate thread using asyncio.

        Args:
            config (RunnableConfig): The config to associate with the writes.
            writes (List[Tuple[str, Any]]): The writes to save, each as a (channel, value) pair.
            task_id (str): Identifier for the task creating the writes.
        """
        return await asyncio.get_running_loop().run_in_executor(
            None, self.put_writes, config, writes, task_id
        )

    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
        if current is None:
            current_v = 0
        elif isinstance(current, int):
            current_v = current
        else:
            current_v = int(current.split(".")[0])
        next_v = current_v + 1
        next_h = random.random()
        return f"{next_v:032}.{next_h:016}"

config_specs: list[ConfigurableFieldSpec] property

Define the configuration options for the checkpoint saver.

Returns:

• list[ConfigurableFieldSpec] –

  list[ConfigurableFieldSpec]: List of configuration field specs.

get(config: RunnableConfig) -> Optional[Checkpoint]

Fetch a checkpoint using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[Checkpoint] –

  Optional[Checkpoint]: The requested checkpoint, or None if not found.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := self.get_tuple(config):
        return value.checkpoint

aget(config: RunnableConfig) -> Optional[Checkpoint] async

Asynchronously fetch a checkpoint using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[Checkpoint] –

  Optional[Checkpoint]: The requested checkpoint, or None if not found.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Asynchronously fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := await self.aget_tuple(config):
        return value.checkpoint

get_tuple(config: RunnableConfig) -> Optional[CheckpointTuple]

Get a checkpoint tuple from the in-memory storage.

This method retrieves a checkpoint tuple from the in-memory storage based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

• config (RunnableConfig) –

  The config to use for retrieving the checkpoint.

Returns:

• Optional[CheckpointTuple] –

  Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

Source code in libs/checkpoint/langgraph/checkpoint/memory/__init__.py

def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Get a checkpoint tuple from the in-memory storage.

    This method retrieves a checkpoint tuple from the in-memory storage based on the
    provided config. If the config contains a "checkpoint_id" key, the checkpoint with
    the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint
    for the given thread ID is retrieved.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
    """
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
    if checkpoint_id := get_checkpoint_id(config):
        if saved := self.storage[thread_id][checkpoint_ns].get(checkpoint_id):
            checkpoint, metadata, parent_checkpoint_id = saved
            writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
            if parent_checkpoint_id:
                sends = [
                    w[2]
                    for w in self.writes[
                        (thread_id, checkpoint_ns, parent_checkpoint_id)
                    ].values()
                    if w[1] == TASKS
                ]
            else:
                sends = []
            return CheckpointTuple(
                config=config,
                checkpoint={
                    **self.serde.loads_typed(checkpoint),
                    "pending_sends": [self.serde.loads_typed(s) for s in sends],
                },
                metadata=self.serde.loads_typed(metadata),
                pending_writes=[
                    (id, c, self.serde.loads_typed(v)) for id, c, v in writes
                ],
                parent_config={
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": parent_checkpoint_id,
                    }
                }
                if parent_checkpoint_id
                else None,
            )
    else:
        if checkpoints := self.storage[thread_id][checkpoint_ns]:
            checkpoint_id = max(checkpoints.keys())
            checkpoint, metadata, parent_checkpoint_id = checkpoints[checkpoint_id]
            writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
            if parent_checkpoint_id:
                sends = [
                    w[2]
                    for w in self.writes[
                        (thread_id, checkpoint_ns, parent_checkpoint_id)
                    ].values()
                    if w[1] == TASKS
                ]
            else:
                sends = []
            return CheckpointTuple(
                config={
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": checkpoint_id,
                    }
                },
                checkpoint={
                    **self.serde.loads_typed(checkpoint),
                    "pending_sends": [self.serde.loads_typed(s) for s in sends],
                },
                metadata=self.serde.loads_typed(metadata),
                pending_writes=[
                    (id, c, self.serde.loads_typed(v)) for id, c, v in writes
                ],
                parent_config={
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": parent_checkpoint_id,
                    }
                }
                if parent_checkpoint_id
                else None,
            )

list(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> Iterator[CheckpointTuple]

List checkpoints from the in-memory storage.

This method retrieves a list of checkpoint tuples from the in-memory storage based on the provided criteria.

Parameters:

• config (Optional[RunnableConfig]) –

  Base configuration for filtering checkpoints.

• filter (Optional[Dict[str, Any]], default: None ) –

  Additional filtering criteria for metadata.

• before (Optional[RunnableConfig], default: None ) –

  List checkpoints created before this configuration.

• limit (Optional[int], default: None ) –

  Maximum number of checkpoints to return.

Yields:

• CheckpointTuple –

  Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.

Source code in libs/checkpoint/langgraph/checkpoint/memory/__init__.py

def list(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
    """List checkpoints from the in-memory storage.

    This method retrieves a list of checkpoint tuples from the in-memory storage based
    on the provided criteria.

    Args:
        config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
        filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
        before (Optional[RunnableConfig]): List checkpoints created before this configuration.
        limit (Optional[int]): Maximum number of checkpoints to return.

    Yields:
        Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.
    """
    thread_ids = (config["configurable"]["thread_id"],) if config else self.storage
    config_checkpoint_ns = (
        config["configurable"].get("checkpoint_ns") if config else None
    )
    config_checkpoint_id = get_checkpoint_id(config) if config else None
    for thread_id in thread_ids:
        for checkpoint_ns in self.storage[thread_id].keys():
            if (
                config_checkpoint_ns is not None
                and checkpoint_ns != config_checkpoint_ns
            ):
                continue

            for checkpoint_id, (
                checkpoint,
                metadata_b,
                parent_checkpoint_id,
            ) in sorted(
                self.storage[thread_id][checkpoint_ns].items(),
                key=lambda x: x[0],
                reverse=True,
            ):
                # filter by checkpoint ID from config
                if config_checkpoint_id and checkpoint_id != config_checkpoint_id:
                    continue

                # filter by checkpoint ID from `before` config
                if (
                    before
                    and (before_checkpoint_id := get_checkpoint_id(before))
                    and checkpoint_id >= before_checkpoint_id
                ):
                    continue

                # filter by metadata
                metadata = self.serde.loads_typed(metadata_b)
                if filter and not all(
                    query_value == metadata.get(query_key)
                    for query_key, query_value in filter.items()
                ):
                    continue

                # limit search results
                if limit is not None and limit <= 0:
                    break
                elif limit is not None:
                    limit -= 1

                writes = self.writes[
                    (thread_id, checkpoint_ns, checkpoint_id)
                ].values()

                if parent_checkpoint_id:
                    sends = [
                        w[2]
                        for w in self.writes[
                            (thread_id, checkpoint_ns, parent_checkpoint_id)
                        ].values()
                        if w[1] == TASKS
                    ]
                else:
                    sends = []

                yield CheckpointTuple(
                    config={
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": checkpoint_id,
                        }
                    },
                    checkpoint={
                        **self.serde.loads_typed(checkpoint),
                        "pending_sends": [self.serde.loads_typed(s) for s in sends],
                    },
                    metadata=metadata,
                    parent_config={
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": parent_checkpoint_id,
                        }
                    }
                    if parent_checkpoint_id
                    else None,
                    pending_writes=[
                        (id, c, self.serde.loads_typed(v)) for id, c, v in writes
                    ],
                )

put(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig

Save a checkpoint to the in-memory storage.

This method saves a checkpoint to the in-memory storage. The checkpoint is associated with the provided config.

Parameters:

• config (RunnableConfig) –

  The config to associate with the checkpoint.

• checkpoint (Checkpoint) –

  The checkpoint to save.

• metadata (CheckpointMetadata) –

  Additional metadata to save with the checkpoint.

• new_versions (dict) –

  New versions as of this write

Returns:

• RunnableConfig ( RunnableConfig ) –

  The updated config containing the saved checkpoint's timestamp.

Source code in libs/checkpoint/langgraph/checkpoint/memory/__init__.py

def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Save a checkpoint to the in-memory storage.

    This method saves a checkpoint to the in-memory storage. The checkpoint is associated
    with the provided config.

    Args:
        config (RunnableConfig): The config to associate with the checkpoint.
        checkpoint (Checkpoint): The checkpoint to save.
        metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
        new_versions (dict): New versions as of this write

    Returns:
        RunnableConfig: The updated config containing the saved checkpoint's timestamp.
    """
    c = checkpoint.copy()
    c.pop("pending_sends")  # type: ignore[misc]
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"]["checkpoint_ns"]
    self.storage[thread_id][checkpoint_ns].update(
        {
            checkpoint["id"]: (
                self.serde.dumps_typed(c),
                self.serde.dumps_typed(metadata),
                config["configurable"].get("checkpoint_id"),  # parent
            )
        }
    )
    return {
        "configurable": {
            "thread_id": thread_id,
            "checkpoint_ns": checkpoint_ns,
            "checkpoint_id": checkpoint["id"],
        }
    }

put_writes(config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str) -> None

Save a list of writes to the in-memory storage.

This method saves a list of writes to the in-memory storage. The writes are associated with the provided config.

Parameters:

• config (RunnableConfig) –

  The config to associate with the writes.

• writes (list[tuple[str, Any]]) –

  The writes to save.

• task_id (str) –

  Identifier for the task creating the writes.

Returns:

• RunnableConfig ( None ) –

  The updated config containing the saved writes' timestamp.

Source code in libs/checkpoint/langgraph/checkpoint/memory/__init__.py

def put_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
) -> None:
    """Save a list of writes to the in-memory storage.

    This method saves a list of writes to the in-memory storage. The writes are associated
    with the provided config.

    Args:
        config (RunnableConfig): The config to associate with the writes.
        writes (list[tuple[str, Any]]): The writes to save.
        task_id (str): Identifier for the task creating the writes.

    Returns:
        RunnableConfig: The updated config containing the saved writes' timestamp.
    """
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
    checkpoint_id = config["configurable"]["checkpoint_id"]
    outer_key = (thread_id, checkpoint_ns, checkpoint_id)
    outer_writes_ = self.writes.get(outer_key)
    for idx, (c, v) in enumerate(writes):
        inner_key = (task_id, WRITES_IDX_MAP.get(c, idx))
        if inner_key[1] >= 0 and outer_writes_ and inner_key in outer_writes_:
            continue

        self.writes[outer_key][inner_key] = (task_id, c, self.serde.dumps_typed(v))

aget_tuple(config: RunnableConfig) -> Optional[CheckpointTuple] async

Asynchronous version of get_tuple.

This method is an asynchronous wrapper around get_tuple that runs the synchronous method in a separate thread using asyncio.

Parameters:

• config (RunnableConfig) –

  The config to use for retrieving the checkpoint.

Returns:

• Optional[CheckpointTuple] –

  Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

Source code in libs/checkpoint/langgraph/checkpoint/memory/__init__.py

async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Asynchronous version of get_tuple.

    This method is an asynchronous wrapper around get_tuple that runs the synchronous
    method in a separate thread using asyncio.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
    """
    return await asyncio.get_running_loop().run_in_executor(
        None, self.get_tuple, config
    )

alist(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> AsyncIterator[CheckpointTuple] async

Asynchronous version of list.

This method is an asynchronous wrapper around list that runs the synchronous method in a separate thread using asyncio.

Parameters:

• config (RunnableConfig) –

  The config to use for listing the checkpoints.

Yields:

• AsyncIterator[CheckpointTuple] –

  AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples.

Source code in libs/checkpoint/langgraph/checkpoint/memory/__init__.py

async def alist(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> AsyncIterator[CheckpointTuple]:
    """Asynchronous version of list.

    This method is an asynchronous wrapper around list that runs the synchronous
    method in a separate thread using asyncio.

    Args:
        config (RunnableConfig): The config to use for listing the checkpoints.

    Yields:
        AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples.
    """
    loop = asyncio.get_running_loop()
    iter = await loop.run_in_executor(
        None,
        partial(
            self.list,
            before=before,
            limit=limit,
            filter=filter,
        ),
        config,
    )
    while True:
        # handling StopIteration exception inside coroutine won't work
        # as expected, so using next() with default value to break the loop
        if item := await loop.run_in_executor(None, next, iter, None):
            yield item
        else:
            break

aput(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig async

Asynchronous version of put.

Parameters:

• config (RunnableConfig) –

  The config to associate with the checkpoint.

• checkpoint (Checkpoint) –

  The checkpoint to save.

• metadata (CheckpointMetadata) –

  Additional metadata to save with the checkpoint.

• new_versions (dict) –

  New versions as of this write

Returns:

• RunnableConfig ( RunnableConfig ) –

  The updated config containing the saved checkpoint's timestamp.

Source code in libs/checkpoint/langgraph/checkpoint/memory/__init__.py

async def aput(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Asynchronous version of put.

    Args:
        config (RunnableConfig): The config to associate with the checkpoint.
        checkpoint (Checkpoint): The checkpoint to save.
        metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
        new_versions (dict): New versions as of this write

    Returns:
        RunnableConfig: The updated config containing the saved checkpoint's timestamp.
    """
    return await asyncio.get_running_loop().run_in_executor(
        None, self.put, config, checkpoint, metadata, new_versions
    )

aput_writes(config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str) -> None async

Asynchronous version of put_writes.

This method is an asynchronous wrapper around put_writes that runs the synchronous method in a separate thread using asyncio.

Parameters:

• config (RunnableConfig) –

  The config to associate with the writes.

• writes (List[Tuple[str, Any]]) –

  The writes to save, each as a (channel, value) pair.

• task_id (str) –

  Identifier for the task creating the writes.

Source code in libs/checkpoint/langgraph/checkpoint/memory/__init__.py

async def aput_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
) -> None:
    """Asynchronous version of put_writes.

    This method is an asynchronous wrapper around put_writes that runs the synchronous
    method in a separate thread using asyncio.

    Args:
        config (RunnableConfig): The config to associate with the writes.
        writes (List[Tuple[str, Any]]): The writes to save, each as a (channel, value) pair.
        task_id (str): Identifier for the task creating the writes.
    """
    return await asyncio.get_running_loop().run_in_executor(
        None, self.put_writes, config, writes, task_id
    )

PersistentDict

Bases: defaultdict

Persistent dictionary with an API compatible with shelve and anydbm.

The dict is kept in memory, so the dictionary operations run as fast as a regular dictionary.

Write to disk is delayed until close or sync (similar to gdbm's fast mode).

Input file format is automatically discovered. Output file format is selectable between pickle, json, and csv. All three serialization formats are backed by fast C implementations.

Adapted from https://code.activestate.com/recipes/576642-persistent-dict-with-multiple-standard-file-format/

Source code in libs/checkpoint/langgraph/checkpoint/memory/__init__.py

class PersistentDict(defaultdict):
    """Persistent dictionary with an API compatible with shelve and anydbm.

    The dict is kept in memory, so the dictionary operations run as fast as
    a regular dictionary.

    Write to disk is delayed until close or sync (similar to gdbm's fast mode).

    Input file format is automatically discovered.
    Output file format is selectable between pickle, json, and csv.
    All three serialization formats are backed by fast C implementations.

    Adapted from https://code.activestate.com/recipes/576642-persistent-dict-with-multiple-standard-file-format/

    """

    def __init__(self, *args: Any, filename: str, **kwds: Any) -> None:
        self.flag = "c"  # r=readonly, c=create, or n=new
        self.mode = None  # None or an octal triple like 0644
        self.format = "pickle"  # 'csv', 'json', or 'pickle'
        self.filename = filename
        super().__init__(*args, **kwds)

    def sync(self) -> None:
        "Write dict to disk"
        if self.flag == "r":
            return
        tempname = self.filename + ".tmp"
        fileobj = open(tempname, "wb" if self.format == "pickle" else "w")
        try:
            self.dump(fileobj)
        except Exception:
            os.remove(tempname)
            raise
        finally:
            fileobj.close()
        shutil.move(tempname, self.filename)  # atomic commit
        if self.mode is not None:
            os.chmod(self.filename, self.mode)

    def close(self) -> None:
        self.sync()
        self.clear()

    def __enter__(self) -> "PersistentDict":
        return self

    def __exit__(self, *exc_info: Any) -> None:
        self.close()

    def dump(self, fileobj: Any) -> None:
        if self.format == "pickle":
            pickle.dump(dict(self), fileobj, 2)
        else:
            raise NotImplementedError("Unknown format: " + repr(self.format))

    def load(self) -> None:
        # try formats from most restrictive to least restrictive
        if self.flag == "n":
            return
        with open(self.filename, "rb" if self.format == "pickle" else "r") as fileobj:
            for loader in (pickle.load,):
                fileobj.seek(0)
                try:
                    return self.update(loader(fileobj))
                except EOFError:
                    return
                except Exception:
                    logging.error(f"Failed to load file: {fileobj.name}")
                    raise
            raise ValueError("File not in a supported f ormat")

sync() -> None

Write dict to disk

Source code in libs/checkpoint/langgraph/checkpoint/memory/__init__.py

def sync(self) -> None:
    "Write dict to disk"
    if self.flag == "r":
        return
    tempname = self.filename + ".tmp"
    fileobj = open(tempname, "wb" if self.format == "pickle" else "w")
    try:
        self.dump(fileobj)
    except Exception:
        os.remove(tempname)
        raise
    finally:
        fileobj.close()
    shutil.move(tempname, self.filename)  # atomic commit
    if self.mode is not None:
        os.chmod(self.filename, self.mode)

SqliteSaver

Bases: BaseCheckpointSaver[str]

A checkpoint saver that stores checkpoints in a SQLite database.

Note

This class is meant for lightweight, synchronous use cases (demos and small projects) and does not scale to multiple threads. For a similar sqlite saver with async support, consider using AsyncSqliteSaver.

Parameters:

• conn (Connection) –

  The SQLite database connection.

• serde (Optional[SerializerProtocol], default: None ) –

  The serializer to use for serializing and deserializing checkpoints. Defaults to JsonPlusSerializerCompat.

Examples:

>>> import sqlite3
>>> 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")
>>> conn = sqlite3.connect("checkpoints.sqlite")
>>> memory = SqliteSaver(conn)
>>> graph = builder.compile(checkpointer=memory)
>>> config = {"configurable": {"thread_id": "1"}}
>>> graph.get_state(config)
>>> result = graph.invoke(3, config)
>>> graph.get_state(config)
StateSnapshot(values=4, next=(), config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '0c62ca34-ac19-445d-bbb0-5b4984975b2a'}}, parent_config=None)

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

class SqliteSaver(BaseCheckpointSaver[str]):
    """A checkpoint saver that stores checkpoints in a SQLite database.

    Note:
        This class is meant for lightweight, synchronous use cases
        (demos and small projects) and does not
        scale to multiple threads.
        For a similar sqlite saver with `async` support,
        consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].

    Args:
        conn (sqlite3.Connection): The SQLite database connection.
        serde (Optional[SerializerProtocol]): The serializer to use for serializing and deserializing checkpoints. Defaults to JsonPlusSerializerCompat.

    Examples:

        >>> import sqlite3
        >>> 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")
        >>> conn = sqlite3.connect("checkpoints.sqlite")
        >>> memory = SqliteSaver(conn)
        >>> graph = builder.compile(checkpointer=memory)
        >>> config = {"configurable": {"thread_id": "1"}}
        >>> graph.get_state(config)
        >>> result = graph.invoke(3, config)
        >>> graph.get_state(config)
        StateSnapshot(values=4, next=(), config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '0c62ca34-ac19-445d-bbb0-5b4984975b2a'}}, parent_config=None)
    """  # noqa

    conn: sqlite3.Connection
    is_setup: bool

    def __init__(
        self,
        conn: sqlite3.Connection,
        *,
        serde: Optional[SerializerProtocol] = None,
    ) -> None:
        super().__init__(serde=serde)
        self.jsonplus_serde = JsonPlusSerializer()
        self.conn = conn
        self.is_setup = False
        self.lock = threading.Lock()

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

        Args:
            conn_string (str): The SQLite connection string.

        Yields:
            SqliteSaver: A new SqliteSaver instance.

        Examples:

            In memory:

                with SqliteSaver.from_conn_string(":memory:") as memory:
                    ...

            To disk:

                with SqliteSaver.from_conn_string("checkpoints.sqlite") as memory:
                    ...
        """
        with closing(
            sqlite3.connect(
                conn_string,
                # https://ricardoanderegg.com/posts/python-sqlite-thread-safety/
                check_same_thread=False,
            )
        ) as conn:
            yield SqliteSaver(conn)

    def setup(self) -> None:
        """Set up the checkpoint database.

        This method creates the necessary tables in the SQLite database if they don't
        already exist. It is called automatically when needed and should not be called
        directly by the user.
        """
        if self.is_setup:
            return

        self.conn.executescript(
            """
            PRAGMA journal_mode=WAL;
            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)
            );
            """
        )

        self.is_setup = True

    @contextmanager
    def cursor(self, transaction: bool = True) -> Iterator[sqlite3.Cursor]:
        """Get a cursor for the SQLite database.

        This method returns a cursor for the SQLite database. It is used internally
        by the SqliteSaver and should not be called directly by the user.

        Args:
            transaction (bool): Whether to commit the transaction when the cursor is closed. Defaults to True.

        Yields:
            sqlite3.Cursor: A cursor for the SQLite database.
        """
        with self.lock:
            self.setup()
            cur = self.conn.cursor()
            try:
                yield cur
            finally:
                if transaction:
                    self.conn.commit()
                cur.close()

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the database.

        This method retrieves a checkpoint tuple from the SQLite database based on the
        provided config. If the config contains a "checkpoint_id" key, the checkpoint with
        the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

        Examples:

            Basic:
            >>> config = {"configurable": {"thread_id": "1"}}
            >>> checkpoint_tuple = memory.get_tuple(config)
            >>> print(checkpoint_tuple)
            CheckpointTuple(...)

            With checkpoint ID:

            >>> config = {
            ...    "configurable": {
            ...        "thread_id": "1",
            ...        "checkpoint_ns": "",
            ...        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
            ...    }
            ... }
            >>> checkpoint_tuple = memory.get_tuple(config)
            >>> print(checkpoint_tuple)
            CheckpointTuple(...)
        """  # noqa
        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
        with self.cursor(transaction=False) as cur:
            # find the latest checkpoint for the thread_id
            if checkpoint_id := get_checkpoint_id(config):
                cur.execute(
                    "SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
                    (
                        str(config["configurable"]["thread_id"]),
                        checkpoint_ns,
                        checkpoint_id,
                    ),
                )
            else:
                cur.execute(
                    "SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? ORDER BY checkpoint_id DESC LIMIT 1",
                    (str(config["configurable"]["thread_id"]), checkpoint_ns),
                )
            # if a checkpoint is found, return it
            if value := cur.fetchone():
                (
                    thread_id,
                    checkpoint_id,
                    parent_checkpoint_id,
                    type,
                    checkpoint,
                    metadata,
                ) = value
                if not get_checkpoint_id(config):
                    config = {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": checkpoint_id,
                        }
                    }
                # find any pending writes
                cur.execute(
                    "SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
                    (
                        str(config["configurable"]["thread_id"]),
                        checkpoint_ns,
                        str(config["configurable"]["checkpoint_id"]),
                    ),
                )
                # deserialize the checkpoint and metadata
                return CheckpointTuple(
                    config,
                    self.serde.loads_typed((type, checkpoint)),
                    self.jsonplus_serde.loads(metadata) if metadata is not None else {},
                    (
                        {
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": parent_checkpoint_id,
                            }
                        }
                        if parent_checkpoint_id
                        else None
                    ),
                    [
                        (task_id, channel, self.serde.loads_typed((type, value)))
                        for task_id, channel, type, value in cur
                    ],
                )

    def list(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
        """List checkpoints from the database.

        This method retrieves a list of checkpoint tuples from the SQLite database based
        on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

        Args:
            config (RunnableConfig): The config to use for listing the checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata. Defaults to None.
            before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
            limit (Optional[int]): The maximum number of checkpoints to return. Defaults to None.

        Yields:
            Iterator[CheckpointTuple]: An iterator of checkpoint tuples.

        Examples:
            >>> from langgraph.checkpoint.sqlite import SqliteSaver
            >>> with SqliteSaver.from_conn_string(":memory:") as memory:
            ... # Run a graph, then list the checkpoints
            >>>     config = {"configurable": {"thread_id": "1"}}
            >>>     checkpoints = list(memory.list(config, limit=2))
            >>> print(checkpoints)
            [CheckpointTuple(...), CheckpointTuple(...)]

            >>> config = {"configurable": {"thread_id": "1"}}
            >>> before = {"configurable": {"checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875"}}
            >>> with SqliteSaver.from_conn_string(":memory:") as memory:
            ... # Run a graph, then list the checkpoints
            >>>     checkpoints = list(memory.list(config, before=before))
            >>> print(checkpoints)
            [CheckpointTuple(...), ...]
        """
        where, param_values = search_where(config, filter, before)
        query = f"""SELECT thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata
        FROM checkpoints
        {where}
        ORDER BY checkpoint_id DESC"""
        if limit:
            query += f" LIMIT {limit}"
        with self.cursor(transaction=False) as cur, closing(self.conn.cursor()) as wcur:
            cur.execute(query, param_values)
            for (
                thread_id,
                checkpoint_ns,
                checkpoint_id,
                parent_checkpoint_id,
                type,
                checkpoint,
                metadata,
            ) in cur:
                wcur.execute(
                    "SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
                    (thread_id, checkpoint_ns, checkpoint_id),
                )
                yield CheckpointTuple(
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": checkpoint_id,
                        }
                    },
                    self.serde.loads_typed((type, checkpoint)),
                    self.jsonplus_serde.loads(metadata) if metadata is not None else {},
                    (
                        {
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": parent_checkpoint_id,
                            }
                        }
                        if parent_checkpoint_id
                        else None
                    ),
                    [
                        (task_id, channel, self.serde.loads_typed((type, value)))
                        for task_id, channel, type, value in wcur
                    ],
                )

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Save a checkpoint to the database.

        This method saves a checkpoint to the SQLite database. The checkpoint is associated
        with the provided config and its parent config (if any).

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.
            metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
            new_versions (ChannelVersions): New channel versions as of this write.

        Returns:
            RunnableConfig: Updated configuration after storing the checkpoint.

        Examples:

            >>> from langgraph.checkpoint.sqlite import SqliteSaver
            >>> with SqliteSaver.from_conn_string(":memory:") as memory:
            >>>     config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
            >>>     checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "channel_values": {"key": "value"}}
            >>>     saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
            >>> print(saved_config)
            {'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}
        """
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"]["checkpoint_ns"]
        type_, serialized_checkpoint = self.serde.dumps_typed(checkpoint)
        serialized_metadata = self.jsonplus_serde.dumps(metadata)
        with self.cursor() as cur:
            cur.execute(
                "INSERT OR REPLACE INTO checkpoints (thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata) VALUES (?, ?, ?, ?, ?, ?, ?)",
                (
                    str(config["configurable"]["thread_id"]),
                    checkpoint_ns,
                    checkpoint["id"],
                    config["configurable"].get("checkpoint_id"),
                    type_,
                    serialized_checkpoint,
                    serialized_metadata,
                ),
            )
        return {
            "configurable": {
                "thread_id": thread_id,
                "checkpoint_ns": checkpoint_ns,
                "checkpoint_id": checkpoint["id"],
            }
        }

    def put_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[Tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Store intermediate writes linked to a checkpoint.

        This method saves intermediate writes associated with a checkpoint to the SQLite database.

        Args:
            config (RunnableConfig): Configuration of the related checkpoint.
            writes (Sequence[Tuple[str, Any]]): List of writes to store, each as (channel, value) pair.
            task_id (str): Identifier for the task creating the writes.
        """
        query = (
            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
            if all(w[0] in WRITES_IDX_MAP for w in writes)
            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
        )
        with self.cursor() as cur:
            cur.executemany(
                query,
                [
                    (
                        str(config["configurable"]["thread_id"]),
                        str(config["configurable"]["checkpoint_ns"]),
                        str(config["configurable"]["checkpoint_id"]),
                        task_id,
                        WRITES_IDX_MAP.get(channel, idx),
                        channel,
                        *self.serde.dumps_typed(value),
                    )
                    for idx, (channel, value) in enumerate(writes)
                ],
            )

    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the database asynchronously.

        Note:
            This async method is not supported by the SqliteSaver class.
            Use get_tuple() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
        """
        raise NotImplementedError(_AIO_ERROR_MSG)

    async def alist(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncIterator[CheckpointTuple]:
        """List checkpoints from the database asynchronously.

        Note:
            This async method is not supported by the SqliteSaver class.
            Use list() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
        """
        raise NotImplementedError(_AIO_ERROR_MSG)
        yield

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Save a checkpoint to the database asynchronously.

        Note:
            This async method is not supported by the SqliteSaver class.
            Use put() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
        """
        raise NotImplementedError(_AIO_ERROR_MSG)

    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
        """Generate the next version ID for a channel.

        This method creates a new version identifier for a channel based on its current version.

        Args:
            current (Optional[str]): The current version identifier of the channel.
            channel (BaseChannel): The channel being versioned.

        Returns:
            str: The next version identifier, which is guaranteed to be monotonically increasing.
        """
        if current is None:
            current_v = 0
        elif isinstance(current, int):
            current_v = current
        else:
            current_v = int(current.split(".")[0])
        next_v = current_v + 1
        next_h = random.random()
        return f"{next_v:032}.{next_h:016}"

config_specs: list[ConfigurableFieldSpec] property

Define the configuration options for the checkpoint saver.

Returns:

• list[ConfigurableFieldSpec] –

  list[ConfigurableFieldSpec]: List of configuration field specs.

get(config: RunnableConfig) -> Optional[Checkpoint]

Fetch a checkpoint using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[Checkpoint] –

  Optional[Checkpoint]: The requested checkpoint, or None if not found.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := self.get_tuple(config):
        return value.checkpoint

aget(config: RunnableConfig) -> Optional[Checkpoint] async

Asynchronously fetch a checkpoint using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[Checkpoint] –

  Optional[Checkpoint]: The requested checkpoint, or None if not found.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Asynchronously fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := await self.aget_tuple(config):
        return value.checkpoint

aput_writes(config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str) -> None async

Asynchronously store intermediate writes linked to a checkpoint.

Parameters:

• config (RunnableConfig) –

  Configuration of the related checkpoint.

• writes (List[Tuple[str, Any]]) –

  List of writes to store.

• task_id (str) –

  Identifier for the task creating the writes.

Raises:

• NotImplementedError –

  Implement this method in your custom checkpoint saver.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def aput_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
) -> None:
    """Asynchronously store intermediate writes linked to a checkpoint.

    Args:
        config (RunnableConfig): Configuration of the related checkpoint.
        writes (List[Tuple[str, Any]]): List of writes to store.
        task_id (str): Identifier for the task creating the writes.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

from_conn_string(conn_string: str) -> Iterator[SqliteSaver] classmethod

Create a new SqliteSaver instance from a connection string.

Parameters:

• conn_string (str) –

  The SQLite connection string.

Yields:

• SqliteSaver ( SqliteSaver ) –

  A new SqliteSaver instance.

Examples:

In memory:

    with SqliteSaver.from_conn_string(":memory:") as memory:
        ...

To disk:

    with SqliteSaver.from_conn_string("checkpoints.sqlite") as memory:
        ...

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

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

    Args:
        conn_string (str): The SQLite connection string.

    Yields:
        SqliteSaver: A new SqliteSaver instance.

    Examples:

        In memory:

            with SqliteSaver.from_conn_string(":memory:") as memory:
                ...

        To disk:

            with SqliteSaver.from_conn_string("checkpoints.sqlite") as memory:
                ...
    """
    with closing(
        sqlite3.connect(
            conn_string,
            # https://ricardoanderegg.com/posts/python-sqlite-thread-safety/
            check_same_thread=False,
        )
    ) as conn:
        yield SqliteSaver(conn)

setup() -> None

Set up the checkpoint database.

This method creates the necessary tables in the SQLite database if they don't already exist. It is called automatically when needed and should not be called directly by the user.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

def setup(self) -> None:
    """Set up the checkpoint database.

    This method creates the necessary tables in the SQLite database if they don't
    already exist. It is called automatically when needed and should not be called
    directly by the user.
    """
    if self.is_setup:
        return

    self.conn.executescript(
        """
        PRAGMA journal_mode=WAL;
        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)
        );
        """
    )

    self.is_setup = True

cursor(transaction: bool = True) -> Iterator[sqlite3.Cursor]

Get a cursor for the SQLite database.

This method returns a cursor for the SQLite database. It is used internally by the SqliteSaver and should not be called directly by the user.

Parameters:

• transaction (bool, default: True ) –

  Whether to commit the transaction when the cursor is closed. Defaults to True.

Yields:

• Cursor –

  sqlite3.Cursor: A cursor for the SQLite database.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

@contextmanager
def cursor(self, transaction: bool = True) -> Iterator[sqlite3.Cursor]:
    """Get a cursor for the SQLite database.

    This method returns a cursor for the SQLite database. It is used internally
    by the SqliteSaver and should not be called directly by the user.

    Args:
        transaction (bool): Whether to commit the transaction when the cursor is closed. Defaults to True.

    Yields:
        sqlite3.Cursor: A cursor for the SQLite database.
    """
    with self.lock:
        self.setup()
        cur = self.conn.cursor()
        try:
            yield cur
        finally:
            if transaction:
                self.conn.commit()
            cur.close()

get_tuple(config: RunnableConfig) -> Optional[CheckpointTuple]

Get a checkpoint tuple from the database.

This method retrieves a checkpoint tuple from the SQLite database based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

• config (RunnableConfig) –

  The config to use for retrieving the checkpoint.

Returns:

• Optional[CheckpointTuple] –

  Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

Examples:

Basic:
>>> config = {"configurable": {"thread_id": "1"}}
>>> checkpoint_tuple = memory.get_tuple(config)
>>> print(checkpoint_tuple)
CheckpointTuple(...)

With checkpoint ID:

>>> config = {
...    "configurable": {
...        "thread_id": "1",
...        "checkpoint_ns": "",
...        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
...    }
... }
>>> checkpoint_tuple = memory.get_tuple(config)
>>> print(checkpoint_tuple)
CheckpointTuple(...)

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Get a checkpoint tuple from the database.

    This method retrieves a checkpoint tuple from the SQLite database based on the
    provided config. If the config contains a "checkpoint_id" key, the checkpoint with
    the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint
    for the given thread ID is retrieved.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

    Examples:

        Basic:
        >>> config = {"configurable": {"thread_id": "1"}}
        >>> checkpoint_tuple = memory.get_tuple(config)
        >>> print(checkpoint_tuple)
        CheckpointTuple(...)

        With checkpoint ID:

        >>> config = {
        ...    "configurable": {
        ...        "thread_id": "1",
        ...        "checkpoint_ns": "",
        ...        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        ...    }
        ... }
        >>> checkpoint_tuple = memory.get_tuple(config)
        >>> print(checkpoint_tuple)
        CheckpointTuple(...)
    """  # noqa
    checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
    with self.cursor(transaction=False) as cur:
        # find the latest checkpoint for the thread_id
        if checkpoint_id := get_checkpoint_id(config):
            cur.execute(
                "SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
                (
                    str(config["configurable"]["thread_id"]),
                    checkpoint_ns,
                    checkpoint_id,
                ),
            )
        else:
            cur.execute(
                "SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? ORDER BY checkpoint_id DESC LIMIT 1",
                (str(config["configurable"]["thread_id"]), checkpoint_ns),
            )
        # if a checkpoint is found, return it
        if value := cur.fetchone():
            (
                thread_id,
                checkpoint_id,
                parent_checkpoint_id,
                type,
                checkpoint,
                metadata,
            ) = value
            if not get_checkpoint_id(config):
                config = {
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": checkpoint_id,
                    }
                }
            # find any pending writes
            cur.execute(
                "SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
                (
                    str(config["configurable"]["thread_id"]),
                    checkpoint_ns,
                    str(config["configurable"]["checkpoint_id"]),
                ),
            )
            # deserialize the checkpoint and metadata
            return CheckpointTuple(
                config,
                self.serde.loads_typed((type, checkpoint)),
                self.jsonplus_serde.loads(metadata) if metadata is not None else {},
                (
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": parent_checkpoint_id,
                        }
                    }
                    if parent_checkpoint_id
                    else None
                ),
                [
                    (task_id, channel, self.serde.loads_typed((type, value)))
                    for task_id, channel, type, value in cur
                ],
            )

list(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> Iterator[CheckpointTuple]

List checkpoints from the database.

This method retrieves a list of checkpoint tuples from the SQLite database based on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

Parameters:

• config (RunnableConfig) –

  The config to use for listing the checkpoints.

• filter (Optional[Dict[str, Any]], default: None ) –

  Additional filtering criteria for metadata. Defaults to None.

• before (Optional[RunnableConfig], default: None ) –

  If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.

• limit (Optional[int], default: None ) –

  The maximum number of checkpoints to return. Defaults to None.

Yields:

• CheckpointTuple –

  Iterator[CheckpointTuple]: An iterator of checkpoint tuples.

Examples:

>>> from langgraph.checkpoint.sqlite import SqliteSaver
>>> with SqliteSaver.from_conn_string(":memory:") as memory:
... # Run a graph, then list the checkpoints
>>>     config = {"configurable": {"thread_id": "1"}}
>>>     checkpoints = list(memory.list(config, limit=2))
>>> print(checkpoints)
[CheckpointTuple(...), CheckpointTuple(...)]

>>> config = {"configurable": {"thread_id": "1"}}
>>> before = {"configurable": {"checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875"}}
>>> with SqliteSaver.from_conn_string(":memory:") as memory:
... # Run a graph, then list the checkpoints
>>>     checkpoints = list(memory.list(config, before=before))
>>> print(checkpoints)
[CheckpointTuple(...), ...]

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

def list(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
    """List checkpoints from the database.

    This method retrieves a list of checkpoint tuples from the SQLite database based
    on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

    Args:
        config (RunnableConfig): The config to use for listing the checkpoints.
        filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata. Defaults to None.
        before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
        limit (Optional[int]): The maximum number of checkpoints to return. Defaults to None.

    Yields:
        Iterator[CheckpointTuple]: An iterator of checkpoint tuples.

    Examples:
        >>> from langgraph.checkpoint.sqlite import SqliteSaver
        >>> with SqliteSaver.from_conn_string(":memory:") as memory:
        ... # Run a graph, then list the checkpoints
        >>>     config = {"configurable": {"thread_id": "1"}}
        >>>     checkpoints = list(memory.list(config, limit=2))
        >>> print(checkpoints)
        [CheckpointTuple(...), CheckpointTuple(...)]

        >>> config = {"configurable": {"thread_id": "1"}}
        >>> before = {"configurable": {"checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875"}}
        >>> with SqliteSaver.from_conn_string(":memory:") as memory:
        ... # Run a graph, then list the checkpoints
        >>>     checkpoints = list(memory.list(config, before=before))
        >>> print(checkpoints)
        [CheckpointTuple(...), ...]
    """
    where, param_values = search_where(config, filter, before)
    query = f"""SELECT thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata
    FROM checkpoints
    {where}
    ORDER BY checkpoint_id DESC"""
    if limit:
        query += f" LIMIT {limit}"
    with self.cursor(transaction=False) as cur, closing(self.conn.cursor()) as wcur:
        cur.execute(query, param_values)
        for (
            thread_id,
            checkpoint_ns,
            checkpoint_id,
            parent_checkpoint_id,
            type,
            checkpoint,
            metadata,
        ) in cur:
            wcur.execute(
                "SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
                (thread_id, checkpoint_ns, checkpoint_id),
            )
            yield CheckpointTuple(
                {
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": checkpoint_id,
                    }
                },
                self.serde.loads_typed((type, checkpoint)),
                self.jsonplus_serde.loads(metadata) if metadata is not None else {},
                (
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": parent_checkpoint_id,
                        }
                    }
                    if parent_checkpoint_id
                    else None
                ),
                [
                    (task_id, channel, self.serde.loads_typed((type, value)))
                    for task_id, channel, type, value in wcur
                ],
            )

put(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig

Save a checkpoint to the database.

This method saves a checkpoint to the SQLite database. The checkpoint is associated with the provided config and its parent config (if any).

Parameters:

• config (RunnableConfig) –

  The config to associate with the checkpoint.

• checkpoint (Checkpoint) –

  The checkpoint to save.

• metadata (CheckpointMetadata) –

  Additional metadata to save with the checkpoint.

• new_versions (ChannelVersions) –

  New channel versions as of this write.

Returns:

• RunnableConfig ( RunnableConfig ) –

  Updated configuration after storing the checkpoint.

Examples:

>>> from langgraph.checkpoint.sqlite import SqliteSaver
>>> with SqliteSaver.from_conn_string(":memory:") as memory:
>>>     config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
>>>     checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "channel_values": {"key": "value"}}
>>>     saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
>>> print(saved_config)
{'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Save a checkpoint to the database.

    This method saves a checkpoint to the SQLite database. The checkpoint is associated
    with the provided config and its parent config (if any).

    Args:
        config (RunnableConfig): The config to associate with the checkpoint.
        checkpoint (Checkpoint): The checkpoint to save.
        metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
        new_versions (ChannelVersions): New channel versions as of this write.

    Returns:
        RunnableConfig: Updated configuration after storing the checkpoint.

    Examples:

        >>> from langgraph.checkpoint.sqlite import SqliteSaver
        >>> with SqliteSaver.from_conn_string(":memory:") as memory:
        >>>     config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
        >>>     checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "channel_values": {"key": "value"}}
        >>>     saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
        >>> print(saved_config)
        {'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}
    """
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"]["checkpoint_ns"]
    type_, serialized_checkpoint = self.serde.dumps_typed(checkpoint)
    serialized_metadata = self.jsonplus_serde.dumps(metadata)
    with self.cursor() as cur:
        cur.execute(
            "INSERT OR REPLACE INTO checkpoints (thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (
                str(config["configurable"]["thread_id"]),
                checkpoint_ns,
                checkpoint["id"],
                config["configurable"].get("checkpoint_id"),
                type_,
                serialized_checkpoint,
                serialized_metadata,
            ),
        )
    return {
        "configurable": {
            "thread_id": thread_id,
            "checkpoint_ns": checkpoint_ns,
            "checkpoint_id": checkpoint["id"],
        }
    }

put_writes(config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str) -> None

Store intermediate writes linked to a checkpoint.

This method saves intermediate writes associated with a checkpoint to the SQLite database.

Parameters:

• config (RunnableConfig) –

  Configuration of the related checkpoint.

• writes (Sequence[Tuple[str, Any]]) –

  List of writes to store, each as (channel, value) pair.

• task_id (str) –

  Identifier for the task creating the writes.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

def put_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
) -> None:
    """Store intermediate writes linked to a checkpoint.

    This method saves intermediate writes associated with a checkpoint to the SQLite database.

    Args:
        config (RunnableConfig): Configuration of the related checkpoint.
        writes (Sequence[Tuple[str, Any]]): List of writes to store, each as (channel, value) pair.
        task_id (str): Identifier for the task creating the writes.
    """
    query = (
        "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
        if all(w[0] in WRITES_IDX_MAP for w in writes)
        else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
    )
    with self.cursor() as cur:
        cur.executemany(
            query,
            [
                (
                    str(config["configurable"]["thread_id"]),
                    str(config["configurable"]["checkpoint_ns"]),
                    str(config["configurable"]["checkpoint_id"]),
                    task_id,
                    WRITES_IDX_MAP.get(channel, idx),
                    channel,
                    *self.serde.dumps_typed(value),
                )
                for idx, (channel, value) in enumerate(writes)
            ],
        )

aget_tuple(config: RunnableConfig) -> Optional[CheckpointTuple] async

Get a checkpoint tuple from the database asynchronously.

Note

This async method is not supported by the SqliteSaver class. Use get_tuple() instead, or consider using AsyncSqliteSaver.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Get a checkpoint tuple from the database asynchronously.

    Note:
        This async method is not supported by the SqliteSaver class.
        Use get_tuple() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
    """
    raise NotImplementedError(_AIO_ERROR_MSG)

alist(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> AsyncIterator[CheckpointTuple] async

List checkpoints from the database asynchronously.

Note

This async method is not supported by the SqliteSaver class. Use list() instead, or consider using AsyncSqliteSaver.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

async def alist(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> AsyncIterator[CheckpointTuple]:
    """List checkpoints from the database asynchronously.

    Note:
        This async method is not supported by the SqliteSaver class.
        Use list() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
    """
    raise NotImplementedError(_AIO_ERROR_MSG)
    yield

aput(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig async

Save a checkpoint to the database asynchronously.

Note

This async method is not supported by the SqliteSaver class. Use put() instead, or consider using AsyncSqliteSaver.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

async def aput(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Save a checkpoint to the database asynchronously.

    Note:
        This async method is not supported by the SqliteSaver class.
        Use put() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
    """
    raise NotImplementedError(_AIO_ERROR_MSG)

get_next_version(current: Optional[str], channel: ChannelProtocol) -> str

Generate the next version ID for a channel.

This method creates a new version identifier for a channel based on its current version.

Parameters:

• current (Optional[str]) –

  The current version identifier of the channel.

• channel (BaseChannel) –

  The channel being versioned.

Returns:

• str ( str ) –

  The next version identifier, which is guaranteed to be monotonically increasing.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
    """Generate the next version ID for a channel.

    This method creates a new version identifier for a channel based on its current version.

    Args:
        current (Optional[str]): The current version identifier of the channel.
        channel (BaseChannel): The channel being versioned.

    Returns:
        str: The next version identifier, which is guaranteed to be monotonically increasing.
    """
    if current is None:
        current_v = 0
    elif isinstance(current, int):
        current_v = current
    else:
        current_v = int(current.split(".")[0])
    next_v = current_v + 1
    next_h = random.random()
    return f"{next_v:032}.{next_h:016}"

AsyncSqliteSaver

Bases: BaseCheckpointSaver[str]

An asynchronous checkpoint saver that stores checkpoints in a SQLite database.

This class provides an asynchronous interface for saving and retrieving checkpoints using a SQLite database. It's designed for use in asynchronous environments and offers better performance for I/O-bound operations compared to synchronous alternatives.

Attributes:

• conn (Connection) –

  The asynchronous SQLite database connection.

• serde (SerializerProtocol) –

  The serializer used for encoding/decoding checkpoints.

Tip

Requires the aiosqlite package. Install it with pip install aiosqlite.

Warning

While this class supports asynchronous checkpointing, it is not recommended for production workloads due to limitations in SQLite's write performance. For production use, consider a more robust database like PostgreSQL.

Tip

Remember to close the database connection after executing your code, otherwise, you may see the graph "hang" after execution (since the program will not exit until the connection is closed).

The easiest way is to use the async with statement as shown in the examples.

async with AsyncSqliteSaver.from_conn_string("checkpoints.sqlite") as saver:
    # Your code here
    graph = builder.compile(checkpointer=saver)
    config = {"configurable": {"thread_id": "thread-1"}}
    async for event in graph.astream_events(..., config, version="v1"):
        print(event)

Examples:

Usage within StateGraph:

>>> import asyncio
>>>
>>> from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
>>> 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")
>>> async with AsyncSqliteSaver.from_conn_string("checkpoints.db") as memory:
>>>     graph = builder.compile(checkpointer=memory)
>>>     coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
>>>     print(asyncio.run(coro))
Output: 2

Raw usage:

>>> import asyncio
>>> import aiosqlite
>>> from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
>>>
>>> async def main():
>>>     async with aiosqlite.connect("checkpoints.db") as conn:
...         saver = AsyncSqliteSaver(conn)
...         config = {"configurable": {"thread_id": "1"}}
...         checkpoint = {"ts": "2023-05-03T10:00:00Z", "data": {"key": "value"}}
...         saved_config = await saver.aput(config, checkpoint, {}, {})
...         print(saved_config)
>>> asyncio.run(main())
{"configurable": {"thread_id": "1", "checkpoint_id": "0c62ca34-ac19-445d-bbb0-5b4984975b2a"}}

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

class AsyncSqliteSaver(BaseCheckpointSaver[str]):
    """An asynchronous checkpoint saver that stores checkpoints in a SQLite database.

    This class provides an asynchronous interface for saving and retrieving checkpoints
    using a SQLite database. It's designed for use in asynchronous environments and
    offers better performance for I/O-bound operations compared to synchronous alternatives.

    Attributes:
        conn (aiosqlite.Connection): The asynchronous SQLite database connection.
        serde (SerializerProtocol): The serializer used for encoding/decoding checkpoints.

    Tip:
        Requires the [aiosqlite](https://pypi.org/project/aiosqlite/) package.
        Install it with `pip install aiosqlite`.

    Warning:
        While this class supports asynchronous checkpointing, it is not recommended
        for production workloads due to limitations in SQLite's write performance.
        For production use, consider a more robust database like PostgreSQL.

    Tip:
        Remember to **close the database connection** after executing your code,
        otherwise, you may see the graph "hang" after execution (since the program
        will not exit until the connection is closed).

        The easiest way is to use the `async with` statement as shown in the examples.

        ```python
        async with AsyncSqliteSaver.from_conn_string("checkpoints.sqlite") as saver:
            # Your code here
            graph = builder.compile(checkpointer=saver)
            config = {"configurable": {"thread_id": "thread-1"}}
            async for event in graph.astream_events(..., config, version="v1"):
                print(event)
        ```

    Examples:
        Usage within StateGraph:

        ```pycon
        >>> import asyncio
        >>>
        >>> from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
        >>> 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")
        >>> async with AsyncSqliteSaver.from_conn_string("checkpoints.db") as memory:
        >>>     graph = builder.compile(checkpointer=memory)
        >>>     coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
        >>>     print(asyncio.run(coro))
        Output: 2
        ```
        Raw usage:

        ```pycon
        >>> import asyncio
        >>> import aiosqlite
        >>> from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
        >>>
        >>> async def main():
        >>>     async with aiosqlite.connect("checkpoints.db") as conn:
        ...         saver = AsyncSqliteSaver(conn)
        ...         config = {"configurable": {"thread_id": "1"}}
        ...         checkpoint = {"ts": "2023-05-03T10:00:00Z", "data": {"key": "value"}}
        ...         saved_config = await saver.aput(config, checkpoint, {}, {})
        ...         print(saved_config)
        >>> asyncio.run(main())
        {"configurable": {"thread_id": "1", "checkpoint_id": "0c62ca34-ac19-445d-bbb0-5b4984975b2a"}}
        ```
    """

    lock: asyncio.Lock
    is_setup: bool

    def __init__(
        self,
        conn: aiosqlite.Connection,
        *,
        serde: Optional[SerializerProtocol] = None,
    ):
        super().__init__(serde=serde)
        self.jsonplus_serde = JsonPlusSerializer()
        self.conn = conn
        self.lock = asyncio.Lock()
        self.loop = asyncio.get_running_loop()
        self.is_setup = False

    @classmethod
    @asynccontextmanager
    async def from_conn_string(
        cls, conn_string: str
    ) -> AsyncIterator["AsyncSqliteSaver"]:
        """Create a new AsyncSqliteSaver instance from a connection string.

        Args:
            conn_string (str): The SQLite connection string.

        Yields:
            AsyncSqliteSaver: A new AsyncSqliteSaver instance.
        """
        async with aiosqlite.connect(conn_string) as conn:
            yield AsyncSqliteSaver(conn)

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the database.

        This method retrieves a checkpoint tuple from the SQLite database based on the
        provided config. If the config contains a "checkpoint_id" key, the checkpoint with
        the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        try:
            # check if we are in the main thread, only bg threads can block
            # we don't check in other methods to avoid the overhead
            if asyncio.get_running_loop() is self.loop:
                raise asyncio.InvalidStateError(
                    "Synchronous calls to AsyncSqliteSaver are only allowed from a "
                    "different thread. From the main thread, use the async interface."
                    "For example, use `await checkpointer.aget_tuple(...)` or `await "
                    "graph.ainvoke(...)`."
                )
        except RuntimeError:
            pass
        return asyncio.run_coroutine_threadsafe(
            self.aget_tuple(config), self.loop
        ).result()

    def list(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
        """List checkpoints from the database asynchronously.

        This method retrieves a list of checkpoint tuples from the SQLite database based
        on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

        Args:
            config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
            before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
            limit (Optional[int]): Maximum number of checkpoints to return.

        Yields:
            Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.
        """
        aiter_ = self.alist(config, filter=filter, before=before, limit=limit)
        while True:
            try:
                yield asyncio.run_coroutine_threadsafe(
                    anext(aiter_),
                    self.loop,
                ).result()
            except StopAsyncIteration:
                break

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Save a checkpoint to the database.

        This method saves a checkpoint to the SQLite database. The checkpoint is associated
        with the provided config and its parent config (if any).

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.
            metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
            new_versions (ChannelVersions): New channel versions as of this write.

        Returns:
            RunnableConfig: Updated configuration after storing the checkpoint.
        """
        return asyncio.run_coroutine_threadsafe(
            self.aput(config, checkpoint, metadata, new_versions), self.loop
        ).result()

    def put_writes(
        self, config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str
    ) -> None:
        return asyncio.run_coroutine_threadsafe(
            self.aput_writes(config, writes, task_id), self.loop
        ).result()

    async def setup(self) -> None:
        """Set up the checkpoint database asynchronously.

        This method creates the necessary tables in the SQLite database if they don't
        already exist. It is called automatically when needed and should not be called
        directly by the user.
        """
        async with self.lock:
            if self.is_setup:
                return
            if not self.conn.is_alive():
                await self.conn
            async with self.conn.executescript(
                """
                PRAGMA journal_mode=WAL;
                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)
                );
                """
            ):
                await self.conn.commit()

            self.is_setup = True

    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the database asynchronously.

        This method retrieves a checkpoint tuple from the SQLite database based on the
        provided config. If the config contains a "checkpoint_id" key, the checkpoint with
        the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        await self.setup()
        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
        async with self.lock, self.conn.cursor() as cur:
            # find the latest checkpoint for the thread_id
            if checkpoint_id := get_checkpoint_id(config):
                await cur.execute(
                    "SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
                    (
                        str(config["configurable"]["thread_id"]),
                        checkpoint_ns,
                        checkpoint_id,
                    ),
                )
            else:
                await cur.execute(
                    "SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? ORDER BY checkpoint_id DESC LIMIT 1",
                    (str(config["configurable"]["thread_id"]), checkpoint_ns),
                )
            # if a checkpoint is found, return it
            if value := await cur.fetchone():
                (
                    thread_id,
                    checkpoint_id,
                    parent_checkpoint_id,
                    type,
                    checkpoint,
                    metadata,
                ) = value
                if not get_checkpoint_id(config):
                    config = {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": checkpoint_id,
                        }
                    }
                # find any pending writes
                await cur.execute(
                    "SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
                    (
                        str(config["configurable"]["thread_id"]),
                        checkpoint_ns,
                        str(config["configurable"]["checkpoint_id"]),
                    ),
                )
                # deserialize the checkpoint and metadata
                return CheckpointTuple(
                    config,
                    self.serde.loads_typed((type, checkpoint)),
                    self.jsonplus_serde.loads(metadata) if metadata is not None else {},
                    (
                        {
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": parent_checkpoint_id,
                            }
                        }
                        if parent_checkpoint_id
                        else None
                    ),
                    [
                        (task_id, channel, self.serde.loads_typed((type, value)))
                        async for task_id, channel, type, value in cur
                    ],
                )

    async def alist(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncIterator[CheckpointTuple]:
        """List checkpoints from the database asynchronously.

        This method retrieves a list of checkpoint tuples from the SQLite database based
        on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

        Args:
            config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
            before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
            limit (Optional[int]): Maximum number of checkpoints to return.

        Yields:
            AsyncIterator[CheckpointTuple]: An asynchronous iterator of matching checkpoint tuples.
        """
        await self.setup()
        where, params = search_where(config, filter, before)
        query = f"""SELECT thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata
        FROM checkpoints
        {where}
        ORDER BY checkpoint_id DESC"""
        if limit:
            query += f" LIMIT {limit}"
        async with self.lock, self.conn.execute(
            query, params
        ) as cur, self.conn.cursor() as wcur:
            async for (
                thread_id,
                checkpoint_ns,
                checkpoint_id,
                parent_checkpoint_id,
                type,
                checkpoint,
                metadata,
            ) in cur:
                await wcur.execute(
                    "SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
                    (thread_id, checkpoint_ns, checkpoint_id),
                )
                yield CheckpointTuple(
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": checkpoint_id,
                        }
                    },
                    self.serde.loads_typed((type, checkpoint)),
                    self.jsonplus_serde.loads(metadata) if metadata is not None else {},
                    (
                        {
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": parent_checkpoint_id,
                            }
                        }
                        if parent_checkpoint_id
                        else None
                    ),
                    [
                        (task_id, channel, self.serde.loads_typed((type, value)))
                        async for task_id, channel, type, value in wcur
                    ],
                )

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Save a checkpoint to the database asynchronously.

        This method saves a checkpoint to the SQLite database. The checkpoint is associated
        with the provided config and its parent config (if any).

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.
            metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
            new_versions (ChannelVersions): New channel versions as of this write.

        Returns:
            RunnableConfig: Updated configuration after storing the checkpoint.
        """
        await self.setup()
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"]["checkpoint_ns"]
        type_, serialized_checkpoint = self.serde.dumps_typed(checkpoint)
        serialized_metadata = self.jsonplus_serde.dumps(metadata)
        async with self.lock, self.conn.execute(
            "INSERT OR REPLACE INTO checkpoints (thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (
                str(config["configurable"]["thread_id"]),
                checkpoint_ns,
                checkpoint["id"],
                config["configurable"].get("checkpoint_id"),
                type_,
                serialized_checkpoint,
                serialized_metadata,
            ),
        ):
            await self.conn.commit()
        return {
            "configurable": {
                "thread_id": thread_id,
                "checkpoint_ns": checkpoint_ns,
                "checkpoint_id": checkpoint["id"],
            }
        }

    async def aput_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[Tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Store intermediate writes linked to a checkpoint asynchronously.

        This method saves intermediate writes associated with a checkpoint to the database.

        Args:
            config (RunnableConfig): Configuration of the related checkpoint.
            writes (Sequence[Tuple[str, Any]]): List of writes to store, each as (channel, value) pair.
            task_id (str): Identifier for the task creating the writes.
        """
        query = (
            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
            if all(w[0] in WRITES_IDX_MAP for w in writes)
            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
        )
        await self.setup()
        async with self.lock, self.conn.cursor() as cur:
            await cur.executemany(
                query,
                [
                    (
                        str(config["configurable"]["thread_id"]),
                        str(config["configurable"]["checkpoint_ns"]),
                        str(config["configurable"]["checkpoint_id"]),
                        task_id,
                        WRITES_IDX_MAP.get(channel, idx),
                        channel,
                        *self.serde.dumps_typed(value),
                    )
                    for idx, (channel, value) in enumerate(writes)
                ],
            )

    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
        """Generate the next version ID for a channel.

        This method creates a new version identifier for a channel based on its current version.

        Args:
            current (Optional[str]): The current version identifier of the channel.
            channel (BaseChannel): The channel being versioned.

        Returns:
            str: The next version identifier, which is guaranteed to be monotonically increasing.
        """
        if current is None:
            current_v = 0
        elif isinstance(current, int):
            current_v = current
        else:
            current_v = int(current.split(".")[0])
        next_v = current_v + 1
        next_h = random.random()
        return f"{next_v:032}.{next_h:016}"

config_specs: list[ConfigurableFieldSpec] property

Define the configuration options for the checkpoint saver.

Returns:

• list[ConfigurableFieldSpec] –

  list[ConfigurableFieldSpec]: List of configuration field specs.

get(config: RunnableConfig) -> Optional[Checkpoint]

Fetch a checkpoint using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[Checkpoint] –

  Optional[Checkpoint]: The requested checkpoint, or None if not found.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := self.get_tuple(config):
        return value.checkpoint

aget(config: RunnableConfig) -> Optional[Checkpoint] async

Asynchronously fetch a checkpoint using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[Checkpoint] –

  Optional[Checkpoint]: The requested checkpoint, or None if not found.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Asynchronously fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := await self.aget_tuple(config):
        return value.checkpoint

from_conn_string(conn_string: str) -> AsyncIterator[AsyncSqliteSaver] async classmethod

Create a new AsyncSqliteSaver instance from a connection string.

Parameters:

• conn_string (str) –

  The SQLite connection string.

Yields:

• AsyncSqliteSaver ( AsyncIterator[AsyncSqliteSaver] ) –

  A new AsyncSqliteSaver instance.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

@classmethod
@asynccontextmanager
async def from_conn_string(
    cls, conn_string: str
) -> AsyncIterator["AsyncSqliteSaver"]:
    """Create a new AsyncSqliteSaver instance from a connection string.

    Args:
        conn_string (str): The SQLite connection string.

    Yields:
        AsyncSqliteSaver: A new AsyncSqliteSaver instance.
    """
    async with aiosqlite.connect(conn_string) as conn:
        yield AsyncSqliteSaver(conn)

get_tuple(config: RunnableConfig) -> Optional[CheckpointTuple]

Get a checkpoint tuple from the database.

This method retrieves a checkpoint tuple from the SQLite database based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

• config (RunnableConfig) –

  The config to use for retrieving the checkpoint.

Returns:

• Optional[CheckpointTuple] –

  Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Get a checkpoint tuple from the database.

    This method retrieves a checkpoint tuple from the SQLite database based on the
    provided config. If the config contains a "checkpoint_id" key, the checkpoint with
    the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint
    for the given thread ID is retrieved.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
    """
    try:
        # check if we are in the main thread, only bg threads can block
        # we don't check in other methods to avoid the overhead
        if asyncio.get_running_loop() is self.loop:
            raise asyncio.InvalidStateError(
                "Synchronous calls to AsyncSqliteSaver are only allowed from a "
                "different thread. From the main thread, use the async interface."
                "For example, use `await checkpointer.aget_tuple(...)` or `await "
                "graph.ainvoke(...)`."
            )
    except RuntimeError:
        pass
    return asyncio.run_coroutine_threadsafe(
        self.aget_tuple(config), self.loop
    ).result()

list(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> Iterator[CheckpointTuple]

List checkpoints from the database asynchronously.

This method retrieves a list of checkpoint tuples from the SQLite database based on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

Parameters:

• config (Optional[RunnableConfig]) –

  Base configuration for filtering checkpoints.

• filter (Optional[Dict[str, Any]], default: None ) –

  Additional filtering criteria for metadata.

• before (Optional[RunnableConfig], default: None ) –

  If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.

• limit (Optional[int], default: None ) –

  Maximum number of checkpoints to return.

Yields:

• CheckpointTuple –

  Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

def list(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
    """List checkpoints from the database asynchronously.

    This method retrieves a list of checkpoint tuples from the SQLite database based
    on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

    Args:
        config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
        filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
        before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
        limit (Optional[int]): Maximum number of checkpoints to return.

    Yields:
        Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.
    """
    aiter_ = self.alist(config, filter=filter, before=before, limit=limit)
    while True:
        try:
            yield asyncio.run_coroutine_threadsafe(
                anext(aiter_),
                self.loop,
            ).result()
        except StopAsyncIteration:
            break

put(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig

Save a checkpoint to the database.

This method saves a checkpoint to the SQLite database. The checkpoint is associated with the provided config and its parent config (if any).

Parameters:

• config (RunnableConfig) –

  The config to associate with the checkpoint.

• checkpoint (Checkpoint) –

  The checkpoint to save.

• metadata (CheckpointMetadata) –

  Additional metadata to save with the checkpoint.

• new_versions (ChannelVersions) –

  New channel versions as of this write.

Returns:

• RunnableConfig ( RunnableConfig ) –

  Updated configuration after storing the checkpoint.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Save a checkpoint to the database.

    This method saves a checkpoint to the SQLite database. The checkpoint is associated
    with the provided config and its parent config (if any).

    Args:
        config (RunnableConfig): The config to associate with the checkpoint.
        checkpoint (Checkpoint): The checkpoint to save.
        metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
        new_versions (ChannelVersions): New channel versions as of this write.

    Returns:
        RunnableConfig: Updated configuration after storing the checkpoint.
    """
    return asyncio.run_coroutine_threadsafe(
        self.aput(config, checkpoint, metadata, new_versions), self.loop
    ).result()

setup() -> None async

Set up the checkpoint database asynchronously.

This method creates the necessary tables in the SQLite database if they don't already exist. It is called automatically when needed and should not be called directly by the user.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

async def setup(self) -> None:
    """Set up the checkpoint database asynchronously.

    This method creates the necessary tables in the SQLite database if they don't
    already exist. It is called automatically when needed and should not be called
    directly by the user.
    """
    async with self.lock:
        if self.is_setup:
            return
        if not self.conn.is_alive():
            await self.conn
        async with self.conn.executescript(
            """
            PRAGMA journal_mode=WAL;
            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)
            );
            """
        ):
            await self.conn.commit()

        self.is_setup = True

aget_tuple(config: RunnableConfig) -> Optional[CheckpointTuple] async

Get a checkpoint tuple from the database asynchronously.

This method retrieves a checkpoint tuple from the SQLite database based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

• config (RunnableConfig) –

  The config to use for retrieving the checkpoint.

Returns:

• Optional[CheckpointTuple] –

  Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Get a checkpoint tuple from the database asynchronously.

    This method retrieves a checkpoint tuple from the SQLite database based on the
    provided config. If the config contains a "checkpoint_id" key, the checkpoint with
    the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint
    for the given thread ID is retrieved.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
    """
    await self.setup()
    checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
    async with self.lock, self.conn.cursor() as cur:
        # find the latest checkpoint for the thread_id
        if checkpoint_id := get_checkpoint_id(config):
            await cur.execute(
                "SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
                (
                    str(config["configurable"]["thread_id"]),
                    checkpoint_ns,
                    checkpoint_id,
                ),
            )
        else:
            await cur.execute(
                "SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? ORDER BY checkpoint_id DESC LIMIT 1",
                (str(config["configurable"]["thread_id"]), checkpoint_ns),
            )
        # if a checkpoint is found, return it
        if value := await cur.fetchone():
            (
                thread_id,
                checkpoint_id,
                parent_checkpoint_id,
                type,
                checkpoint,
                metadata,
            ) = value
            if not get_checkpoint_id(config):
                config = {
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": checkpoint_id,
                    }
                }
            # find any pending writes
            await cur.execute(
                "SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
                (
                    str(config["configurable"]["thread_id"]),
                    checkpoint_ns,
                    str(config["configurable"]["checkpoint_id"]),
                ),
            )
            # deserialize the checkpoint and metadata
            return CheckpointTuple(
                config,
                self.serde.loads_typed((type, checkpoint)),
                self.jsonplus_serde.loads(metadata) if metadata is not None else {},
                (
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": parent_checkpoint_id,
                        }
                    }
                    if parent_checkpoint_id
                    else None
                ),
                [
                    (task_id, channel, self.serde.loads_typed((type, value)))
                    async for task_id, channel, type, value in cur
                ],
            )

alist(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> AsyncIterator[CheckpointTuple] async

List checkpoints from the database asynchronously.

This method retrieves a list of checkpoint tuples from the SQLite database based on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

Parameters:

• config (Optional[RunnableConfig]) –

  Base configuration for filtering checkpoints.

• filter (Optional[Dict[str, Any]], default: None ) –

  Additional filtering criteria for metadata.

• before (Optional[RunnableConfig], default: None ) –

  If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.

• limit (Optional[int], default: None ) –

  Maximum number of checkpoints to return.

Yields:

• AsyncIterator[CheckpointTuple] –

  AsyncIterator[CheckpointTuple]: An asynchronous iterator of matching checkpoint tuples.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

async def alist(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> AsyncIterator[CheckpointTuple]:
    """List checkpoints from the database asynchronously.

    This method retrieves a list of checkpoint tuples from the SQLite database based
    on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

    Args:
        config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
        filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
        before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
        limit (Optional[int]): Maximum number of checkpoints to return.

    Yields:
        AsyncIterator[CheckpointTuple]: An asynchronous iterator of matching checkpoint tuples.
    """
    await self.setup()
    where, params = search_where(config, filter, before)
    query = f"""SELECT thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata
    FROM checkpoints
    {where}
    ORDER BY checkpoint_id DESC"""
    if limit:
        query += f" LIMIT {limit}"
    async with self.lock, self.conn.execute(
        query, params
    ) as cur, self.conn.cursor() as wcur:
        async for (
            thread_id,
            checkpoint_ns,
            checkpoint_id,
            parent_checkpoint_id,
            type,
            checkpoint,
            metadata,
        ) in cur:
            await wcur.execute(
                "SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
                (thread_id, checkpoint_ns, checkpoint_id),
            )
            yield CheckpointTuple(
                {
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": checkpoint_id,
                    }
                },
                self.serde.loads_typed((type, checkpoint)),
                self.jsonplus_serde.loads(metadata) if metadata is not None else {},
                (
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": parent_checkpoint_id,
                        }
                    }
                    if parent_checkpoint_id
                    else None
                ),
                [
                    (task_id, channel, self.serde.loads_typed((type, value)))
                    async for task_id, channel, type, value in wcur
                ],
            )

aput(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig async

Save a checkpoint to the database asynchronously.

This method saves a checkpoint to the SQLite database. The checkpoint is associated with the provided config and its parent config (if any).

Parameters:

• config (RunnableConfig) –

  The config to associate with the checkpoint.

• checkpoint (Checkpoint) –

  The checkpoint to save.

• metadata (CheckpointMetadata) –

  Additional metadata to save with the checkpoint.

• new_versions (ChannelVersions) –

  New channel versions as of this write.

Returns:

• RunnableConfig ( RunnableConfig ) –

  Updated configuration after storing the checkpoint.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

async def aput(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Save a checkpoint to the database asynchronously.

    This method saves a checkpoint to the SQLite database. The checkpoint is associated
    with the provided config and its parent config (if any).

    Args:
        config (RunnableConfig): The config to associate with the checkpoint.
        checkpoint (Checkpoint): The checkpoint to save.
        metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
        new_versions (ChannelVersions): New channel versions as of this write.

    Returns:
        RunnableConfig: Updated configuration after storing the checkpoint.
    """
    await self.setup()
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"]["checkpoint_ns"]
    type_, serialized_checkpoint = self.serde.dumps_typed(checkpoint)
    serialized_metadata = self.jsonplus_serde.dumps(metadata)
    async with self.lock, self.conn.execute(
        "INSERT OR REPLACE INTO checkpoints (thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata) VALUES (?, ?, ?, ?, ?, ?, ?)",
        (
            str(config["configurable"]["thread_id"]),
            checkpoint_ns,
            checkpoint["id"],
            config["configurable"].get("checkpoint_id"),
            type_,
            serialized_checkpoint,
            serialized_metadata,
        ),
    ):
        await self.conn.commit()
    return {
        "configurable": {
            "thread_id": thread_id,
            "checkpoint_ns": checkpoint_ns,
            "checkpoint_id": checkpoint["id"],
        }
    }

aput_writes(config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str) -> None async

Store intermediate writes linked to a checkpoint asynchronously.

This method saves intermediate writes associated with a checkpoint to the database.

Parameters:

• config (RunnableConfig) –

  Configuration of the related checkpoint.

• writes (Sequence[Tuple[str, Any]]) –

  List of writes to store, each as (channel, value) pair.

• task_id (str) –

  Identifier for the task creating the writes.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

async def aput_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
) -> None:
    """Store intermediate writes linked to a checkpoint asynchronously.

    This method saves intermediate writes associated with a checkpoint to the database.

    Args:
        config (RunnableConfig): Configuration of the related checkpoint.
        writes (Sequence[Tuple[str, Any]]): List of writes to store, each as (channel, value) pair.
        task_id (str): Identifier for the task creating the writes.
    """
    query = (
        "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
        if all(w[0] in WRITES_IDX_MAP for w in writes)
        else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
    )
    await self.setup()
    async with self.lock, self.conn.cursor() as cur:
        await cur.executemany(
            query,
            [
                (
                    str(config["configurable"]["thread_id"]),
                    str(config["configurable"]["checkpoint_ns"]),
                    str(config["configurable"]["checkpoint_id"]),
                    task_id,
                    WRITES_IDX_MAP.get(channel, idx),
                    channel,
                    *self.serde.dumps_typed(value),
                )
                for idx, (channel, value) in enumerate(writes)
            ],
        )

get_next_version(current: Optional[str], channel: ChannelProtocol) -> str

Generate the next version ID for a channel.

This method creates a new version identifier for a channel based on its current version.

Parameters:

• current (Optional[str]) –

  The current version identifier of the channel.

• channel (BaseChannel) –

  The channel being versioned.

Returns:

• str ( str ) –

  The next version identifier, which is guaranteed to be monotonically increasing.

Source code in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
    """Generate the next version ID for a channel.

    This method creates a new version identifier for a channel based on its current version.

    Args:
        current (Optional[str]): The current version identifier of the channel.
        channel (BaseChannel): The channel being versioned.

    Returns:
        str: The next version identifier, which is guaranteed to be monotonically increasing.
    """
    if current is None:
        current_v = 0
    elif isinstance(current, int):
        current_v = current
    else:
        current_v = int(current.split(".")[0])
    next_v = current_v + 1
    next_h = random.random()
    return f"{next_v:032}.{next_h:016}"

PostgresSaver

Bases: BasePostgresSaver

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py

class PostgresSaver(BasePostgresSaver):
    lock: threading.Lock

    def __init__(
        self,
        conn: Conn,
        pipe: Optional[Pipeline] = None,
        serde: Optional[SerializerProtocol] = None,
    ) -> None:
        super().__init__(serde=serde)
        if isinstance(conn, ConnectionPool) and pipe is not None:
            raise ValueError(
                "Pipeline should be used only with a single Connection, not ConnectionPool."
            )

        self.conn = conn
        self.pipe = pipe
        self.lock = threading.Lock()
        self.supports_pipeline = Capabilities().has_pipeline()

    @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 (str): The Postgres connection info string.
            pipeline (bool): whether to use Pipeline

        Returns:
            PostgresSaver: A new PostgresSaver instance.
        """
        with Connection.connect(
            conn_string, autocommit=True, prepare_threshold=0, row_factory=dict_row
        ) as conn:
            if pipeline:
                with conn.pipeline() as pipe:
                    yield PostgresSaver(conn, pipe)
            else:
                yield PostgresSaver(conn)

    def setup(self) -> None:
        """Set up the checkpoint database asynchronously.

        This method creates the necessary tables in the Postgres database if they don't
        already exist and runs database migrations. It MUST be called directly by the user
        the first time checkpointer is used.
        """
        with self._cursor() as cur:
            try:
                row = cur.execute(
                    "SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
                ).fetchone()
                if row is None:
                    version = -1
                else:
                    version = row["v"]
            except UndefinedTable:
                version = -1
            for v, migration in zip(
                range(version + 1, len(self.MIGRATIONS)),
                self.MIGRATIONS[version + 1 :],
            ):
                cur.execute(migration)
                cur.execute(f"INSERT INTO checkpoint_migrations (v) VALUES ({v})")
        if self.pipe:
            self.pipe.sync()

    def list(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
        """List checkpoints from the database.

        This method retrieves a list of checkpoint tuples from the Postgres database based
        on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

        Args:
            config (RunnableConfig): The config to use for listing the checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata. Defaults to None.
            before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
            limit (Optional[int]): The maximum number of checkpoints to return. Defaults to None.

        Yields:
            Iterator[CheckpointTuple]: An iterator of checkpoint tuples.

        Examples:
            >>> from langgraph.checkpoint.postgres import PostgresSaver
            >>> DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
            >>> with PostgresSaver.from_conn_string(DB_URI) as memory:
            ... # Run a graph, then list the checkpoints
            >>>     config = {"configurable": {"thread_id": "1"}}
            >>>     checkpoints = list(memory.list(config, limit=2))
            >>> print(checkpoints)
            [CheckpointTuple(...), CheckpointTuple(...)]

            >>> config = {"configurable": {"thread_id": "1"}}
            >>> before = {"configurable": {"checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875"}}
            >>> with PostgresSaver.from_conn_string(DB_URI) as memory:
            ... # Run a graph, then list the checkpoints
            >>>     checkpoints = list(memory.list(config, before=before))
            >>> print(checkpoints)
            [CheckpointTuple(...), ...]
        """
        where, args = self._search_where(config, filter, before)
        query = self.SELECT_SQL + where + " ORDER BY checkpoint_id DESC"
        if limit:
            query += f" LIMIT {limit}"
        # if we change this to use .stream() we need to make sure to close the cursor
        with self._cursor() as cur:
            cur.execute(query, args, binary=True)
            for value in cur:
                yield CheckpointTuple(
                    {
                        "configurable": {
                            "thread_id": value["thread_id"],
                            "checkpoint_ns": value["checkpoint_ns"],
                            "checkpoint_id": value["checkpoint_id"],
                        }
                    },
                    self._load_checkpoint(
                        value["checkpoint"],
                        value["channel_values"],
                        value["pending_sends"],
                    ),
                    self._load_metadata(value["metadata"]),
                    (
                        {
                            "configurable": {
                                "thread_id": value["thread_id"],
                                "checkpoint_ns": value["checkpoint_ns"],
                                "checkpoint_id": value["parent_checkpoint_id"],
                            }
                        }
                        if value["parent_checkpoint_id"]
                        else None
                    ),
                    self._load_writes(value["pending_writes"]),
                )

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the database.

        This method retrieves a checkpoint tuple from the Postgres database based on the
        provided config. If the config contains a "checkpoint_id" key, the checkpoint with
        the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

        Examples:

            Basic:
            >>> config = {"configurable": {"thread_id": "1"}}
            >>> checkpoint_tuple = memory.get_tuple(config)
            >>> print(checkpoint_tuple)
            CheckpointTuple(...)

            With timestamp:

            >>> config = {
            ...    "configurable": {
            ...        "thread_id": "1",
            ...        "checkpoint_ns": "",
            ...        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
            ...    }
            ... }
            >>> checkpoint_tuple = memory.get_tuple(config)
            >>> print(checkpoint_tuple)
            CheckpointTuple(...)
        """  # noqa
        thread_id = config["configurable"]["thread_id"]
        checkpoint_id = get_checkpoint_id(config)
        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
        if checkpoint_id:
            args: tuple[Any, ...] = (thread_id, checkpoint_ns, checkpoint_id)
            where = "WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id = %s"
        else:
            args = (thread_id, checkpoint_ns)
            where = "WHERE thread_id = %s AND checkpoint_ns = %s ORDER BY checkpoint_id DESC LIMIT 1"

        with self._cursor() as cur:
            cur.execute(
                self.SELECT_SQL + where,
                args,
                binary=True,
            )

            for value in cur:
                return CheckpointTuple(
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": value["checkpoint_id"],
                        }
                    },
                    self._load_checkpoint(
                        value["checkpoint"],
                        value["channel_values"],
                        value["pending_sends"],
                    ),
                    self._load_metadata(value["metadata"]),
                    (
                        {
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": value["parent_checkpoint_id"],
                            }
                        }
                        if value["parent_checkpoint_id"]
                        else None
                    ),
                    self._load_writes(value["pending_writes"]),
                )

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Save a checkpoint to the database.

        This method saves a checkpoint to the Postgres database. The checkpoint is associated
        with the provided config and its parent config (if any).

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.
            metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
            new_versions (ChannelVersions): New channel versions as of this write.

        Returns:
            RunnableConfig: Updated configuration after storing the checkpoint.

        Examples:

            >>> from langgraph.checkpoint.postgres import PostgresSaver
            >>> DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
            >>> with PostgresSaver.from_conn_string(DB_URI) as memory:
            >>>     config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
            >>>     checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "channel_values": {"key": "value"}}
            >>>     saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
            >>> print(saved_config)
            {'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}
        """
        configurable = config["configurable"].copy()
        thread_id = configurable.pop("thread_id")
        checkpoint_ns = configurable.pop("checkpoint_ns")
        checkpoint_id = configurable.pop(
            "checkpoint_id", configurable.pop("thread_ts", None)
        )

        copy = checkpoint.copy()
        next_config = {
            "configurable": {
                "thread_id": thread_id,
                "checkpoint_ns": checkpoint_ns,
                "checkpoint_id": checkpoint["id"],
            }
        }

        with self._cursor(pipeline=True) as cur:
            cur.executemany(
                self.UPSERT_CHECKPOINT_BLOBS_SQL,
                self._dump_blobs(
                    thread_id,
                    checkpoint_ns,
                    copy.pop("channel_values"),  # type: ignore[misc]
                    new_versions,
                ),
            )
            cur.execute(
                self.UPSERT_CHECKPOINTS_SQL,
                (
                    thread_id,
                    checkpoint_ns,
                    checkpoint["id"],
                    checkpoint_id,
                    Jsonb(self._dump_checkpoint(copy)),
                    self._dump_metadata(metadata),
                ),
            )
        return next_config

    def put_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Store intermediate writes linked to a checkpoint.

        This method saves intermediate writes associated with a checkpoint to the Postgres database.

        Args:
            config (RunnableConfig): Configuration of the related checkpoint.
            writes (List[Tuple[str, Any]]): List of writes to store.
            task_id (str): Identifier for the task creating the writes.
        """
        query = (
            self.UPSERT_CHECKPOINT_WRITES_SQL
            if all(w[0] in WRITES_IDX_MAP for w in writes)
            else self.INSERT_CHECKPOINT_WRITES_SQL
        )
        with self._cursor(pipeline=True) as cur:
            cur.executemany(
                query,
                self._dump_writes(
                    config["configurable"]["thread_id"],
                    config["configurable"]["checkpoint_ns"],
                    config["configurable"]["checkpoint_id"],
                    task_id,
                    writes,
                ),
            )

    @contextmanager
    def _cursor(self, *, pipeline: bool = False) -> Iterator[Cursor[DictRow]]:
        """Create a database cursor as a context manager.

        Args:
            pipeline (bool): whether to use pipeline for the DB operations inside the context manager.
                Will be applied regardless of whether the PostgresSaver instance was initialized with a pipeline.
                If pipeline mode is not supported, will fall back to using transaction context manager.
        """
        with _get_connection(self.conn) as conn:
            if self.pipe:
                # a connection in pipeline mode can be used concurrently
                # in multiple threads/coroutines, but only one cursor can be
                # used at a time
                try:
                    with conn.cursor(binary=True, row_factory=dict_row) as cur:
                        yield cur
                finally:
                    if pipeline:
                        self.pipe.sync()
            elif pipeline:
                # a connection not in pipeline mode can only be used by one
                # thread/coroutine at a time, so we acquire a lock
                if self.supports_pipeline:
                    with self.lock, conn.pipeline(), conn.cursor(
                        binary=True, row_factory=dict_row
                    ) as cur:
                        yield cur
                else:
                    # Use connection's transaction context manager when pipeline mode not supported
                    with self.lock, conn.transaction(), conn.cursor(
                        binary=True, row_factory=dict_row
                    ) as cur:
                        yield cur
            else:
                with self.lock, conn.cursor(binary=True, row_factory=dict_row) as cur:
                    yield cur

config_specs: list[ConfigurableFieldSpec] property

Define the configuration options for the checkpoint saver.

Returns:

• list[ConfigurableFieldSpec] –

  list[ConfigurableFieldSpec]: List of configuration field specs.

get(config: RunnableConfig) -> Optional[Checkpoint]

Fetch a checkpoint using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[Checkpoint] –

  Optional[Checkpoint]: The requested checkpoint, or None if not found.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := self.get_tuple(config):
        return value.checkpoint

aget(config: RunnableConfig) -> Optional[Checkpoint] async

Asynchronously fetch a checkpoint using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[Checkpoint] –

  Optional[Checkpoint]: The requested checkpoint, or None if not found.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Asynchronously fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := await self.aget_tuple(config):
        return value.checkpoint

aget_tuple(config: RunnableConfig) -> Optional[CheckpointTuple] async

Asynchronously fetch a checkpoint tuple using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[CheckpointTuple] –

  Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

Raises:

• NotImplementedError –

  Implement this method in your custom checkpoint saver.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Asynchronously fetch a checkpoint tuple using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

alist(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> AsyncIterator[CheckpointTuple] async

Asynchronously list checkpoints that match the given criteria.

Parameters:

• config (Optional[RunnableConfig]) –

  Base configuration for filtering checkpoints.

• filter (Optional[Dict[str, Any]], default: None ) –

  Additional filtering criteria for metadata.

• before (Optional[RunnableConfig], default: None ) –

  List checkpoints created before this configuration.

• limit (Optional[int], default: None ) –

  Maximum number of checkpoints to return.

Returns:

• AsyncIterator[CheckpointTuple] –

  AsyncIterator[CheckpointTuple]: Async iterator of matching checkpoint tuples.

Raises:

• NotImplementedError –

  Implement this method in your custom checkpoint saver.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def alist(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> AsyncIterator[CheckpointTuple]:
    """Asynchronously list checkpoints that match the given criteria.

    Args:
        config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
        filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
        before (Optional[RunnableConfig]): List checkpoints created before this configuration.
        limit (Optional[int]): Maximum number of checkpoints to return.

    Returns:
        AsyncIterator[CheckpointTuple]: Async iterator of matching checkpoint tuples.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError
    yield

aput(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig async

Asynchronously store a checkpoint with its configuration and metadata.

Parameters:

• config (RunnableConfig) –

  Configuration for the checkpoint.

• checkpoint (Checkpoint) –

  The checkpoint to store.

• metadata (CheckpointMetadata) –

  Additional metadata for the checkpoint.

• new_versions (ChannelVersions) –

  New channel versions as of this write.

Returns:

• RunnableConfig ( RunnableConfig ) –

  Updated configuration after storing the checkpoint.

Raises:

• NotImplementedError –

  Implement this method in your custom checkpoint saver.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def aput(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Asynchronously store a checkpoint with its configuration and metadata.

    Args:
        config (RunnableConfig): Configuration for the checkpoint.
        checkpoint (Checkpoint): The checkpoint to store.
        metadata (CheckpointMetadata): Additional metadata for the checkpoint.
        new_versions (ChannelVersions): New channel versions as of this write.

    Returns:
        RunnableConfig: Updated configuration after storing the checkpoint.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

aput_writes(config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str) -> None async

Asynchronously store intermediate writes linked to a checkpoint.

Parameters:

• config (RunnableConfig) –

  Configuration of the related checkpoint.

• writes (List[Tuple[str, Any]]) –

  List of writes to store.

• task_id (str) –

  Identifier for the task creating the writes.

Raises:

• NotImplementedError –

  Implement this method in your custom checkpoint saver.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def aput_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
) -> None:
    """Asynchronously store intermediate writes linked to a checkpoint.

    Args:
        config (RunnableConfig): Configuration of the related checkpoint.
        writes (List[Tuple[str, Any]]): List of writes to store.
        task_id (str): Identifier for the task creating the writes.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

from_conn_string(conn_string: str, *, pipeline: bool = False) -> Iterator[PostgresSaver] classmethod

Create a new PostgresSaver instance from a connection string.

Parameters:

• conn_string (str) –

  The Postgres connection info string.

• pipeline (bool, default: False ) –

  whether to use Pipeline

Returns:

• PostgresSaver ( Iterator[PostgresSaver] ) –

  A new PostgresSaver instance.

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py

@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 (str): The Postgres connection info string.
        pipeline (bool): whether to use Pipeline

    Returns:
        PostgresSaver: A new PostgresSaver instance.
    """
    with Connection.connect(
        conn_string, autocommit=True, prepare_threshold=0, row_factory=dict_row
    ) as conn:
        if pipeline:
            with conn.pipeline() as pipe:
                yield PostgresSaver(conn, pipe)
        else:
            yield PostgresSaver(conn)

setup() -> None

Set up the checkpoint database asynchronously.

This method creates the necessary tables in the Postgres database if they don't already exist and runs database migrations. It MUST be called directly by the user the first time checkpointer is used.

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py

def setup(self) -> None:
    """Set up the checkpoint database asynchronously.

    This method creates the necessary tables in the Postgres database if they don't
    already exist and runs database migrations. It MUST be called directly by the user
    the first time checkpointer is used.
    """
    with self._cursor() as cur:
        try:
            row = cur.execute(
                "SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
            ).fetchone()
            if row is None:
                version = -1
            else:
                version = row["v"]
        except UndefinedTable:
            version = -1
        for v, migration in zip(
            range(version + 1, len(self.MIGRATIONS)),
            self.MIGRATIONS[version + 1 :],
        ):
            cur.execute(migration)
            cur.execute(f"INSERT INTO checkpoint_migrations (v) VALUES ({v})")
    if self.pipe:
        self.pipe.sync()

list(config: Optional[RunnableConfig], *, filter: Optional[dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> Iterator[CheckpointTuple]

List checkpoints from the database.

This method retrieves a list of checkpoint tuples from the Postgres database based on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

Parameters:

• config (RunnableConfig) –

  The config to use for listing the checkpoints.

• filter (Optional[Dict[str, Any]], default: None ) –

  Additional filtering criteria for metadata. Defaults to None.

• before (Optional[RunnableConfig], default: None ) –

  If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.

• limit (Optional[int], default: None ) –

  The maximum number of checkpoints to return. Defaults to None.

Yields:

• CheckpointTuple –

  Iterator[CheckpointTuple]: An iterator of checkpoint tuples.

Examples:

>>> from langgraph.checkpoint.postgres import PostgresSaver
>>> DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
>>> with PostgresSaver.from_conn_string(DB_URI) as memory:
... # Run a graph, then list the checkpoints
>>>     config = {"configurable": {"thread_id": "1"}}
>>>     checkpoints = list(memory.list(config, limit=2))
>>> print(checkpoints)
[CheckpointTuple(...), CheckpointTuple(...)]

>>> config = {"configurable": {"thread_id": "1"}}
>>> before = {"configurable": {"checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875"}}
>>> with PostgresSaver.from_conn_string(DB_URI) as memory:
... # Run a graph, then list the checkpoints
>>>     checkpoints = list(memory.list(config, before=before))
>>> print(checkpoints)
[CheckpointTuple(...), ...]

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py

def list(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
    """List checkpoints from the database.

    This method retrieves a list of checkpoint tuples from the Postgres database based
    on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

    Args:
        config (RunnableConfig): The config to use for listing the checkpoints.
        filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata. Defaults to None.
        before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
        limit (Optional[int]): The maximum number of checkpoints to return. Defaults to None.

    Yields:
        Iterator[CheckpointTuple]: An iterator of checkpoint tuples.

    Examples:
        >>> from langgraph.checkpoint.postgres import PostgresSaver
        >>> DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
        >>> with PostgresSaver.from_conn_string(DB_URI) as memory:
        ... # Run a graph, then list the checkpoints
        >>>     config = {"configurable": {"thread_id": "1"}}
        >>>     checkpoints = list(memory.list(config, limit=2))
        >>> print(checkpoints)
        [CheckpointTuple(...), CheckpointTuple(...)]

        >>> config = {"configurable": {"thread_id": "1"}}
        >>> before = {"configurable": {"checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875"}}
        >>> with PostgresSaver.from_conn_string(DB_URI) as memory:
        ... # Run a graph, then list the checkpoints
        >>>     checkpoints = list(memory.list(config, before=before))
        >>> print(checkpoints)
        [CheckpointTuple(...), ...]
    """
    where, args = self._search_where(config, filter, before)
    query = self.SELECT_SQL + where + " ORDER BY checkpoint_id DESC"
    if limit:
        query += f" LIMIT {limit}"
    # if we change this to use .stream() we need to make sure to close the cursor
    with self._cursor() as cur:
        cur.execute(query, args, binary=True)
        for value in cur:
            yield CheckpointTuple(
                {
                    "configurable": {
                        "thread_id": value["thread_id"],
                        "checkpoint_ns": value["checkpoint_ns"],
                        "checkpoint_id": value["checkpoint_id"],
                    }
                },
                self._load_checkpoint(
                    value["checkpoint"],
                    value["channel_values"],
                    value["pending_sends"],
                ),
                self._load_metadata(value["metadata"]),
                (
                    {
                        "configurable": {
                            "thread_id": value["thread_id"],
                            "checkpoint_ns": value["checkpoint_ns"],
                            "checkpoint_id": value["parent_checkpoint_id"],
                        }
                    }
                    if value["parent_checkpoint_id"]
                    else None
                ),
                self._load_writes(value["pending_writes"]),
            )

get_tuple(config: RunnableConfig) -> Optional[CheckpointTuple]

Get a checkpoint tuple from the database.

This method retrieves a checkpoint tuple from the Postgres database based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

• config (RunnableConfig) –

  The config to use for retrieving the checkpoint.

Returns:

• Optional[CheckpointTuple] –

  Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

Examples:

Basic:
>>> config = {"configurable": {"thread_id": "1"}}
>>> checkpoint_tuple = memory.get_tuple(config)
>>> print(checkpoint_tuple)
CheckpointTuple(...)

With timestamp:

>>> config = {
...    "configurable": {
...        "thread_id": "1",
...        "checkpoint_ns": "",
...        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
...    }
... }
>>> checkpoint_tuple = memory.get_tuple(config)
>>> print(checkpoint_tuple)
CheckpointTuple(...)

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py

def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Get a checkpoint tuple from the database.

    This method retrieves a checkpoint tuple from the Postgres database based on the
    provided config. If the config contains a "checkpoint_id" key, the checkpoint with
    the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint
    for the given thread ID is retrieved.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

    Examples:

        Basic:
        >>> config = {"configurable": {"thread_id": "1"}}
        >>> checkpoint_tuple = memory.get_tuple(config)
        >>> print(checkpoint_tuple)
        CheckpointTuple(...)

        With timestamp:

        >>> config = {
        ...    "configurable": {
        ...        "thread_id": "1",
        ...        "checkpoint_ns": "",
        ...        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        ...    }
        ... }
        >>> checkpoint_tuple = memory.get_tuple(config)
        >>> print(checkpoint_tuple)
        CheckpointTuple(...)
    """  # noqa
    thread_id = config["configurable"]["thread_id"]
    checkpoint_id = get_checkpoint_id(config)
    checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
    if checkpoint_id:
        args: tuple[Any, ...] = (thread_id, checkpoint_ns, checkpoint_id)
        where = "WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id = %s"
    else:
        args = (thread_id, checkpoint_ns)
        where = "WHERE thread_id = %s AND checkpoint_ns = %s ORDER BY checkpoint_id DESC LIMIT 1"

    with self._cursor() as cur:
        cur.execute(
            self.SELECT_SQL + where,
            args,
            binary=True,
        )

        for value in cur:
            return CheckpointTuple(
                {
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": value["checkpoint_id"],
                    }
                },
                self._load_checkpoint(
                    value["checkpoint"],
                    value["channel_values"],
                    value["pending_sends"],
                ),
                self._load_metadata(value["metadata"]),
                (
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": value["parent_checkpoint_id"],
                        }
                    }
                    if value["parent_checkpoint_id"]
                    else None
                ),
                self._load_writes(value["pending_writes"]),
            )

put(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig

Save a checkpoint to the database.

This method saves a checkpoint to the Postgres database. The checkpoint is associated with the provided config and its parent config (if any).

Parameters:

• config (RunnableConfig) –

  The config to associate with the checkpoint.

• checkpoint (Checkpoint) –

  The checkpoint to save.

• metadata (CheckpointMetadata) –

  Additional metadata to save with the checkpoint.

• new_versions (ChannelVersions) –

  New channel versions as of this write.

Returns:

• RunnableConfig ( RunnableConfig ) –

  Updated configuration after storing the checkpoint.

Examples:

>>> from langgraph.checkpoint.postgres import PostgresSaver
>>> DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
>>> with PostgresSaver.from_conn_string(DB_URI) as memory:
>>>     config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
>>>     checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "channel_values": {"key": "value"}}
>>>     saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
>>> print(saved_config)
{'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py

def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Save a checkpoint to the database.

    This method saves a checkpoint to the Postgres database. The checkpoint is associated
    with the provided config and its parent config (if any).

    Args:
        config (RunnableConfig): The config to associate with the checkpoint.
        checkpoint (Checkpoint): The checkpoint to save.
        metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
        new_versions (ChannelVersions): New channel versions as of this write.

    Returns:
        RunnableConfig: Updated configuration after storing the checkpoint.

    Examples:

        >>> from langgraph.checkpoint.postgres import PostgresSaver
        >>> DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
        >>> with PostgresSaver.from_conn_string(DB_URI) as memory:
        >>>     config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
        >>>     checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "channel_values": {"key": "value"}}
        >>>     saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
        >>> print(saved_config)
        {'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}
    """
    configurable = config["configurable"].copy()
    thread_id = configurable.pop("thread_id")
    checkpoint_ns = configurable.pop("checkpoint_ns")
    checkpoint_id = configurable.pop(
        "checkpoint_id", configurable.pop("thread_ts", None)
    )

    copy = checkpoint.copy()
    next_config = {
        "configurable": {
            "thread_id": thread_id,
            "checkpoint_ns": checkpoint_ns,
            "checkpoint_id": checkpoint["id"],
        }
    }

    with self._cursor(pipeline=True) as cur:
        cur.executemany(
            self.UPSERT_CHECKPOINT_BLOBS_SQL,
            self._dump_blobs(
                thread_id,
                checkpoint_ns,
                copy.pop("channel_values"),  # type: ignore[misc]
                new_versions,
            ),
        )
        cur.execute(
            self.UPSERT_CHECKPOINTS_SQL,
            (
                thread_id,
                checkpoint_ns,
                checkpoint["id"],
                checkpoint_id,
                Jsonb(self._dump_checkpoint(copy)),
                self._dump_metadata(metadata),
            ),
        )
    return next_config

put_writes(config: RunnableConfig, writes: Sequence[tuple[str, Any]], task_id: str) -> None

Store intermediate writes linked to a checkpoint.

This method saves intermediate writes associated with a checkpoint to the Postgres database.

Parameters:

• config (RunnableConfig) –

  Configuration of the related checkpoint.

• writes (List[Tuple[str, Any]]) –

  List of writes to store.

• task_id (str) –

  Identifier for the task creating the writes.

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py

def put_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[tuple[str, Any]],
    task_id: str,
) -> None:
    """Store intermediate writes linked to a checkpoint.

    This method saves intermediate writes associated with a checkpoint to the Postgres database.

    Args:
        config (RunnableConfig): Configuration of the related checkpoint.
        writes (List[Tuple[str, Any]]): List of writes to store.
        task_id (str): Identifier for the task creating the writes.
    """
    query = (
        self.UPSERT_CHECKPOINT_WRITES_SQL
        if all(w[0] in WRITES_IDX_MAP for w in writes)
        else self.INSERT_CHECKPOINT_WRITES_SQL
    )
    with self._cursor(pipeline=True) as cur:
        cur.executemany(
            query,
            self._dump_writes(
                config["configurable"]["thread_id"],
                config["configurable"]["checkpoint_ns"],
                config["configurable"]["checkpoint_id"],
                task_id,
                writes,
            ),
        )

AsyncPostgresSaver

Bases: BasePostgresSaver

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

class AsyncPostgresSaver(BasePostgresSaver):
    lock: asyncio.Lock

    def __init__(
        self,
        conn: Conn,
        pipe: Optional[AsyncPipeline] = None,
        serde: Optional[SerializerProtocol] = None,
    ) -> None:
        super().__init__(serde=serde)
        if isinstance(conn, AsyncConnectionPool) and pipe is not None:
            raise ValueError(
                "Pipeline should be used only with a single AsyncConnection, not AsyncConnectionPool."
            )

        self.conn = conn
        self.pipe = pipe
        self.lock = asyncio.Lock()
        self.loop = asyncio.get_running_loop()
        self.supports_pipeline = Capabilities().has_pipeline()

    @classmethod
    @asynccontextmanager
    async def from_conn_string(
        cls,
        conn_string: str,
        *,
        pipeline: bool = False,
        serde: Optional[SerializerProtocol] = None,
    ) -> AsyncIterator["AsyncPostgresSaver"]:
        """Create a new PostgresSaver instance from a connection string.

        Args:
            conn_string (str): The Postgres connection info string.
            pipeline (bool): whether to use AsyncPipeline

        Returns:
            AsyncPostgresSaver: A new AsyncPostgresSaver instance.
        """
        async with await AsyncConnection.connect(
            conn_string, autocommit=True, prepare_threshold=0, row_factory=dict_row
        ) as conn:
            if pipeline:
                async with conn.pipeline() as pipe:
                    yield AsyncPostgresSaver(conn=conn, pipe=pipe, serde=serde)
            else:
                yield AsyncPostgresSaver(conn=conn, serde=serde)

    async def setup(self) -> None:
        """Set up the checkpoint database asynchronously.

        This method creates the necessary tables in the Postgres database if they don't
        already exist and runs database migrations. It MUST be called directly by the user
        the first time checkpointer is used.
        """
        async with self._cursor() as cur:
            try:
                results = await cur.execute(
                    "SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
                )
                row = await results.fetchone()
                if row is None:
                    version = -1
                else:
                    version = row["v"]
            except UndefinedTable:
                version = -1
            for v, migration in zip(
                range(version + 1, len(self.MIGRATIONS)),
                self.MIGRATIONS[version + 1 :],
            ):
                await cur.execute(migration)
                await cur.execute(f"INSERT INTO checkpoint_migrations (v) VALUES ({v})")
        if self.pipe:
            await self.pipe.sync()

    async def alist(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncIterator[CheckpointTuple]:
        """List checkpoints from the database asynchronously.

        This method retrieves a list of checkpoint tuples from the Postgres database based
        on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

        Args:
            config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
            before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
            limit (Optional[int]): Maximum number of checkpoints to return.

        Yields:
            AsyncIterator[CheckpointTuple]: An asynchronous iterator of matching checkpoint tuples.
        """
        where, args = self._search_where(config, filter, before)
        query = self.SELECT_SQL + where + " ORDER BY checkpoint_id DESC"
        if limit:
            query += f" LIMIT {limit}"
        # if we change this to use .stream() we need to make sure to close the cursor
        async with self._cursor() as cur:
            await cur.execute(query, args, binary=True)
            async for value in cur:
                yield CheckpointTuple(
                    {
                        "configurable": {
                            "thread_id": value["thread_id"],
                            "checkpoint_ns": value["checkpoint_ns"],
                            "checkpoint_id": value["checkpoint_id"],
                        }
                    },
                    await asyncio.to_thread(
                        self._load_checkpoint,
                        value["checkpoint"],
                        value["channel_values"],
                        value["pending_sends"],
                    ),
                    self._load_metadata(value["metadata"]),
                    {
                        "configurable": {
                            "thread_id": value["thread_id"],
                            "checkpoint_ns": value["checkpoint_ns"],
                            "checkpoint_id": value["parent_checkpoint_id"],
                        }
                    }
                    if value["parent_checkpoint_id"]
                    else None,
                    await asyncio.to_thread(self._load_writes, value["pending_writes"]),
                )

    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the database asynchronously.

        This method retrieves a checkpoint tuple from the Postgres database based on the
        provided config. If the config contains a "checkpoint_id" key, the checkpoint with
        the matching thread ID and "checkpoint_id" is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        thread_id = config["configurable"]["thread_id"]
        checkpoint_id = get_checkpoint_id(config)
        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
        if checkpoint_id:
            args: tuple[Any, ...] = (thread_id, checkpoint_ns, checkpoint_id)
            where = "WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id = %s"
        else:
            args = (thread_id, checkpoint_ns)
            where = "WHERE thread_id = %s AND checkpoint_ns = %s ORDER BY checkpoint_id DESC LIMIT 1"

        async with self._cursor() as cur:
            await cur.execute(
                self.SELECT_SQL + where,
                args,
                binary=True,
            )

            async for value in cur:
                return CheckpointTuple(
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": value["checkpoint_id"],
                        }
                    },
                    await asyncio.to_thread(
                        self._load_checkpoint,
                        value["checkpoint"],
                        value["channel_values"],
                        value["pending_sends"],
                    ),
                    self._load_metadata(value["metadata"]),
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": value["parent_checkpoint_id"],
                        }
                    }
                    if value["parent_checkpoint_id"]
                    else None,
                    await asyncio.to_thread(self._load_writes, value["pending_writes"]),
                )

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Save a checkpoint to the database asynchronously.

        This method saves a checkpoint to the Postgres database. The checkpoint is associated
        with the provided config and its parent config (if any).

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.
            metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
            new_versions (ChannelVersions): New channel versions as of this write.

        Returns:
            RunnableConfig: Updated configuration after storing the checkpoint.
        """
        configurable = config["configurable"].copy()
        thread_id = configurable.pop("thread_id")
        checkpoint_ns = configurable.pop("checkpoint_ns")
        checkpoint_id = configurable.pop(
            "checkpoint_id", configurable.pop("thread_ts", None)
        )

        copy = checkpoint.copy()
        next_config = {
            "configurable": {
                "thread_id": thread_id,
                "checkpoint_ns": checkpoint_ns,
                "checkpoint_id": checkpoint["id"],
            }
        }

        async with self._cursor(pipeline=True) as cur:
            await cur.executemany(
                self.UPSERT_CHECKPOINT_BLOBS_SQL,
                await asyncio.to_thread(
                    self._dump_blobs,
                    thread_id,
                    checkpoint_ns,
                    copy.pop("channel_values"),  # type: ignore[misc]
                    new_versions,
                ),
            )
            await cur.execute(
                self.UPSERT_CHECKPOINTS_SQL,
                (
                    thread_id,
                    checkpoint_ns,
                    checkpoint["id"],
                    checkpoint_id,
                    Jsonb(self._dump_checkpoint(copy)),
                    self._dump_metadata(metadata),
                ),
            )
        return next_config

    async def aput_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Store intermediate writes linked to a checkpoint asynchronously.

        This method saves intermediate writes associated with a checkpoint to the database.

        Args:
            config (RunnableConfig): Configuration of the related checkpoint.
            writes (Sequence[Tuple[str, Any]]): List of writes to store, each as (channel, value) pair.
            task_id (str): Identifier for the task creating the writes.
        """
        query = (
            self.UPSERT_CHECKPOINT_WRITES_SQL
            if all(w[0] in WRITES_IDX_MAP for w in writes)
            else self.INSERT_CHECKPOINT_WRITES_SQL
        )
        params = await asyncio.to_thread(
            self._dump_writes,
            config["configurable"]["thread_id"],
            config["configurable"]["checkpoint_ns"],
            config["configurable"]["checkpoint_id"],
            task_id,
            writes,
        )
        async with self._cursor(pipeline=True) as cur:
            await cur.executemany(query, params)

    @asynccontextmanager
    async def _cursor(
        self, *, pipeline: bool = False
    ) -> AsyncIterator[AsyncCursor[DictRow]]:
        """Create a database cursor as a context manager.

        Args:
            pipeline (bool): whether to use pipeline for the DB operations inside the context manager.
                Will be applied regardless of whether the AsyncPostgresSaver instance was initialized with a pipeline.
                If pipeline mode is not supported, will fall back to using transaction context manager.
        """
        async with _get_connection(self.conn) as conn:
            if self.pipe:
                # a connection in pipeline mode can be used concurrently
                # in multiple threads/coroutines, but only one cursor can be
                # used at a time
                try:
                    async with conn.cursor(binary=True, row_factory=dict_row) as cur:
                        yield cur
                finally:
                    if pipeline:
                        await self.pipe.sync()
            elif pipeline:
                # a connection not in pipeline mode can only be used by one
                # thread/coroutine at a time, so we acquire a lock
                if self.supports_pipeline:
                    async with self.lock, conn.pipeline(), conn.cursor(
                        binary=True, row_factory=dict_row
                    ) as cur:
                        yield cur
                else:
                    # Use connection's transaction context manager when pipeline mode not supported
                    async with self.lock, conn.transaction(), conn.cursor(
                        binary=True, row_factory=dict_row
                    ) as cur:
                        yield cur
            else:
                async with self.lock, conn.cursor(
                    binary=True, row_factory=dict_row
                ) as cur:
                    yield cur

    def list(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
        """List checkpoints from the database.

        This method retrieves a list of checkpoint tuples from the Postgres database based
        on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

        Args:
            config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
            before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
            limit (Optional[int]): Maximum number of checkpoints to return.

        Yields:
            Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.
        """
        aiter_ = self.alist(config, filter=filter, before=before, limit=limit)
        while True:
            try:
                yield asyncio.run_coroutine_threadsafe(
                    anext(aiter_),
                    self.loop,
                ).result()
            except StopAsyncIteration:
                break

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the database.

        This method retrieves a checkpoint tuple from the Postgres database based on the
        provided config. If the config contains a "checkpoint_id" key, the checkpoint with
        the matching thread ID and "checkpoint_id" is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        try:
            # check if we are in the main thread, only bg threads can block
            # we don't check in other methods to avoid the overhead
            if asyncio.get_running_loop() is self.loop:
                raise asyncio.InvalidStateError(
                    "Synchronous calls to AsyncPostgresSaver are only allowed from a "
                    "different thread. From the main thread, use the async interface."
                    "For example, use `await checkpointer.aget_tuple(...)` or `await "
                    "graph.ainvoke(...)`."
                )
        except RuntimeError:
            pass
        return asyncio.run_coroutine_threadsafe(
            self.aget_tuple(config), self.loop
        ).result()

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Save a checkpoint to the database.

        This method saves a checkpoint to the Postgres database. The checkpoint is associated
        with the provided config and its parent config (if any).

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.
            metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
            new_versions (ChannelVersions): New channel versions as of this write.

        Returns:
            RunnableConfig: Updated configuration after storing the checkpoint.
        """
        return asyncio.run_coroutine_threadsafe(
            self.aput(config, checkpoint, metadata, new_versions), self.loop
        ).result()

    def put_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Store intermediate writes linked to a checkpoint.

        This method saves intermediate writes associated with a checkpoint to the database.

        Args:
            config (RunnableConfig): Configuration of the related checkpoint.
            writes (Sequence[Tuple[str, Any]]): List of writes to store, each as (channel, value) pair.
            task_id (str): Identifier for the task creating the writes.
        """
        return asyncio.run_coroutine_threadsafe(
            self.aput_writes(config, writes, task_id), self.loop
        ).result()

config_specs: list[ConfigurableFieldSpec] property

Define the configuration options for the checkpoint saver.

Returns:

• list[ConfigurableFieldSpec] –

  list[ConfigurableFieldSpec]: List of configuration field specs.

get(config: RunnableConfig) -> Optional[Checkpoint]

Fetch a checkpoint using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[Checkpoint] –

  Optional[Checkpoint]: The requested checkpoint, or None if not found.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := self.get_tuple(config):
        return value.checkpoint

aget(config: RunnableConfig) -> Optional[Checkpoint] async

Asynchronously fetch a checkpoint using the given configuration.

Parameters:

• config (RunnableConfig) –

  Configuration specifying which checkpoint to retrieve.

Returns:

• Optional[Checkpoint] –

  Optional[Checkpoint]: The requested checkpoint, or None if not found.

Source code in libs/checkpoint/langgraph/checkpoint/base/__init__.py

async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Asynchronously fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := await self.aget_tuple(config):
        return value.checkpoint

from_conn_string(conn_string: str, *, pipeline: bool = False, serde: Optional[SerializerProtocol] = None) -> AsyncIterator[AsyncPostgresSaver] async classmethod

Create a new PostgresSaver instance from a connection string.

Parameters:

• conn_string (str) –

  The Postgres connection info string.

• pipeline (bool, default: False ) –

  whether to use AsyncPipeline

Returns:

• AsyncPostgresSaver ( AsyncIterator[AsyncPostgresSaver] ) –

  A new AsyncPostgresSaver instance.

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

@classmethod
@asynccontextmanager
async def from_conn_string(
    cls,
    conn_string: str,
    *,
    pipeline: bool = False,
    serde: Optional[SerializerProtocol] = None,
) -> AsyncIterator["AsyncPostgresSaver"]:
    """Create a new PostgresSaver instance from a connection string.

    Args:
        conn_string (str): The Postgres connection info string.
        pipeline (bool): whether to use AsyncPipeline

    Returns:
        AsyncPostgresSaver: A new AsyncPostgresSaver instance.
    """
    async with await AsyncConnection.connect(
        conn_string, autocommit=True, prepare_threshold=0, row_factory=dict_row
    ) as conn:
        if pipeline:
            async with conn.pipeline() as pipe:
                yield AsyncPostgresSaver(conn=conn, pipe=pipe, serde=serde)
        else:
            yield AsyncPostgresSaver(conn=conn, serde=serde)

setup() -> None async

Set up the checkpoint database asynchronously.

This method creates the necessary tables in the Postgres database if they don't already exist and runs database migrations. It MUST be called directly by the user the first time checkpointer is used.

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

async def setup(self) -> None:
    """Set up the checkpoint database asynchronously.

    This method creates the necessary tables in the Postgres database if they don't
    already exist and runs database migrations. It MUST be called directly by the user
    the first time checkpointer is used.
    """
    async with self._cursor() as cur:
        try:
            results = await cur.execute(
                "SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
            )
            row = await results.fetchone()
            if row is None:
                version = -1
            else:
                version = row["v"]
        except UndefinedTable:
            version = -1
        for v, migration in zip(
            range(version + 1, len(self.MIGRATIONS)),
            self.MIGRATIONS[version + 1 :],
        ):
            await cur.execute(migration)
            await cur.execute(f"INSERT INTO checkpoint_migrations (v) VALUES ({v})")
    if self.pipe:
        await self.pipe.sync()

alist(config: Optional[RunnableConfig], *, filter: Optional[dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> AsyncIterator[CheckpointTuple] async

List checkpoints from the database asynchronously.

This method retrieves a list of checkpoint tuples from the Postgres database based on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

Parameters:

• config (Optional[RunnableConfig]) –

  Base configuration for filtering checkpoints.

• filter (Optional[Dict[str, Any]], default: None ) –

  Additional filtering criteria for metadata.

• before (Optional[RunnableConfig], default: None ) –

  If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.

• limit (Optional[int], default: None ) –

  Maximum number of checkpoints to return.

Yields:

• AsyncIterator[CheckpointTuple] –

  AsyncIterator[CheckpointTuple]: An asynchronous iterator of matching checkpoint tuples.

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

async def alist(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> AsyncIterator[CheckpointTuple]:
    """List checkpoints from the database asynchronously.

    This method retrieves a list of checkpoint tuples from the Postgres database based
    on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

    Args:
        config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
        filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
        before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
        limit (Optional[int]): Maximum number of checkpoints to return.

    Yields:
        AsyncIterator[CheckpointTuple]: An asynchronous iterator of matching checkpoint tuples.
    """
    where, args = self._search_where(config, filter, before)
    query = self.SELECT_SQL + where + " ORDER BY checkpoint_id DESC"
    if limit:
        query += f" LIMIT {limit}"
    # if we change this to use .stream() we need to make sure to close the cursor
    async with self._cursor() as cur:
        await cur.execute(query, args, binary=True)
        async for value in cur:
            yield CheckpointTuple(
                {
                    "configurable": {
                        "thread_id": value["thread_id"],
                        "checkpoint_ns": value["checkpoint_ns"],
                        "checkpoint_id": value["checkpoint_id"],
                    }
                },
                await asyncio.to_thread(
                    self._load_checkpoint,
                    value["checkpoint"],
                    value["channel_values"],
                    value["pending_sends"],
                ),
                self._load_metadata(value["metadata"]),
                {
                    "configurable": {
                        "thread_id": value["thread_id"],
                        "checkpoint_ns": value["checkpoint_ns"],
                        "checkpoint_id": value["parent_checkpoint_id"],
                    }
                }
                if value["parent_checkpoint_id"]
                else None,
                await asyncio.to_thread(self._load_writes, value["pending_writes"]),
            )

aget_tuple(config: RunnableConfig) -> Optional[CheckpointTuple] async

Get a checkpoint tuple from the database asynchronously.

This method retrieves a checkpoint tuple from the Postgres database based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and "checkpoint_id" is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

• config (RunnableConfig) –

  The config to use for retrieving the checkpoint.

Returns:

• Optional[CheckpointTuple] –

  Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Get a checkpoint tuple from the database asynchronously.

    This method retrieves a checkpoint tuple from the Postgres database based on the
    provided config. If the config contains a "checkpoint_id" key, the checkpoint with
    the matching thread ID and "checkpoint_id" is retrieved. Otherwise, the latest checkpoint
    for the given thread ID is retrieved.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
    """
    thread_id = config["configurable"]["thread_id"]
    checkpoint_id = get_checkpoint_id(config)
    checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
    if checkpoint_id:
        args: tuple[Any, ...] = (thread_id, checkpoint_ns, checkpoint_id)
        where = "WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id = %s"
    else:
        args = (thread_id, checkpoint_ns)
        where = "WHERE thread_id = %s AND checkpoint_ns = %s ORDER BY checkpoint_id DESC LIMIT 1"

    async with self._cursor() as cur:
        await cur.execute(
            self.SELECT_SQL + where,
            args,
            binary=True,
        )

        async for value in cur:
            return CheckpointTuple(
                {
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": value["checkpoint_id"],
                    }
                },
                await asyncio.to_thread(
                    self._load_checkpoint,
                    value["checkpoint"],
                    value["channel_values"],
                    value["pending_sends"],
                ),
                self._load_metadata(value["metadata"]),
                {
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": value["parent_checkpoint_id"],
                    }
                }
                if value["parent_checkpoint_id"]
                else None,
                await asyncio.to_thread(self._load_writes, value["pending_writes"]),
            )

aput(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig async

Save a checkpoint to the database asynchronously.

This method saves a checkpoint to the Postgres database. The checkpoint is associated with the provided config and its parent config (if any).

Parameters:

• config (RunnableConfig) –

  The config to associate with the checkpoint.

• checkpoint (Checkpoint) –

  The checkpoint to save.

• metadata (CheckpointMetadata) –

  Additional metadata to save with the checkpoint.

• new_versions (ChannelVersions) –

  New channel versions as of this write.

Returns:

• RunnableConfig ( RunnableConfig ) –

  Updated configuration after storing the checkpoint.

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

async def aput(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Save a checkpoint to the database asynchronously.

    This method saves a checkpoint to the Postgres database. The checkpoint is associated
    with the provided config and its parent config (if any).

    Args:
        config (RunnableConfig): The config to associate with the checkpoint.
        checkpoint (Checkpoint): The checkpoint to save.
        metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
        new_versions (ChannelVersions): New channel versions as of this write.

    Returns:
        RunnableConfig: Updated configuration after storing the checkpoint.
    """
    configurable = config["configurable"].copy()
    thread_id = configurable.pop("thread_id")
    checkpoint_ns = configurable.pop("checkpoint_ns")
    checkpoint_id = configurable.pop(
        "checkpoint_id", configurable.pop("thread_ts", None)
    )

    copy = checkpoint.copy()
    next_config = {
        "configurable": {
            "thread_id": thread_id,
            "checkpoint_ns": checkpoint_ns,
            "checkpoint_id": checkpoint["id"],
        }
    }

    async with self._cursor(pipeline=True) as cur:
        await cur.executemany(
            self.UPSERT_CHECKPOINT_BLOBS_SQL,
            await asyncio.to_thread(
                self._dump_blobs,
                thread_id,
                checkpoint_ns,
                copy.pop("channel_values"),  # type: ignore[misc]
                new_versions,
            ),
        )
        await cur.execute(
            self.UPSERT_CHECKPOINTS_SQL,
            (
                thread_id,
                checkpoint_ns,
                checkpoint["id"],
                checkpoint_id,
                Jsonb(self._dump_checkpoint(copy)),
                self._dump_metadata(metadata),
            ),
        )
    return next_config

aput_writes(config: RunnableConfig, writes: Sequence[tuple[str, Any]], task_id: str) -> None async

Store intermediate writes linked to a checkpoint asynchronously.

This method saves intermediate writes associated with a checkpoint to the database.

Parameters:

• config (RunnableConfig) –

  Configuration of the related checkpoint.

• writes (Sequence[Tuple[str, Any]]) –

  List of writes to store, each as (channel, value) pair.

• task_id (str) –

  Identifier for the task creating the writes.

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

async def aput_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[tuple[str, Any]],
    task_id: str,
) -> None:
    """Store intermediate writes linked to a checkpoint asynchronously.

    This method saves intermediate writes associated with a checkpoint to the database.

    Args:
        config (RunnableConfig): Configuration of the related checkpoint.
        writes (Sequence[Tuple[str, Any]]): List of writes to store, each as (channel, value) pair.
        task_id (str): Identifier for the task creating the writes.
    """
    query = (
        self.UPSERT_CHECKPOINT_WRITES_SQL
        if all(w[0] in WRITES_IDX_MAP for w in writes)
        else self.INSERT_CHECKPOINT_WRITES_SQL
    )
    params = await asyncio.to_thread(
        self._dump_writes,
        config["configurable"]["thread_id"],
        config["configurable"]["checkpoint_ns"],
        config["configurable"]["checkpoint_id"],
        task_id,
        writes,
    )
    async with self._cursor(pipeline=True) as cur:
        await cur.executemany(query, params)

list(config: Optional[RunnableConfig], *, filter: Optional[dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> Iterator[CheckpointTuple]

List checkpoints from the database.

This method retrieves a list of checkpoint tuples from the Postgres database based on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

Parameters:

• config (Optional[RunnableConfig]) –

  Base configuration for filtering checkpoints.

• filter (Optional[Dict[str, Any]], default: None ) –

  Additional filtering criteria for metadata.

• before (Optional[RunnableConfig], default: None ) –

  If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.

• limit (Optional[int], default: None ) –

  Maximum number of checkpoints to return.

Yields:

• CheckpointTuple –

  Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

def list(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
    """List checkpoints from the database.

    This method retrieves a list of checkpoint tuples from the Postgres database based
    on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

    Args:
        config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
        filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
        before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
        limit (Optional[int]): Maximum number of checkpoints to return.

    Yields:
        Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.
    """
    aiter_ = self.alist(config, filter=filter, before=before, limit=limit)
    while True:
        try:
            yield asyncio.run_coroutine_threadsafe(
                anext(aiter_),
                self.loop,
            ).result()
        except StopAsyncIteration:
            break

get_tuple(config: RunnableConfig) -> Optional[CheckpointTuple]

Get a checkpoint tuple from the database.

This method retrieves a checkpoint tuple from the Postgres database based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and "checkpoint_id" is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

• config (RunnableConfig) –

  The config to use for retrieving the checkpoint.

Returns:

• Optional[CheckpointTuple] –

  Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Get a checkpoint tuple from the database.

    This method retrieves a checkpoint tuple from the Postgres database based on the
    provided config. If the config contains a "checkpoint_id" key, the checkpoint with
    the matching thread ID and "checkpoint_id" is retrieved. Otherwise, the latest checkpoint
    for the given thread ID is retrieved.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
    """
    try:
        # check if we are in the main thread, only bg threads can block
        # we don't check in other methods to avoid the overhead
        if asyncio.get_running_loop() is self.loop:
            raise asyncio.InvalidStateError(
                "Synchronous calls to AsyncPostgresSaver are only allowed from a "
                "different thread. From the main thread, use the async interface."
                "For example, use `await checkpointer.aget_tuple(...)` or `await "
                "graph.ainvoke(...)`."
            )
    except RuntimeError:
        pass
    return asyncio.run_coroutine_threadsafe(
        self.aget_tuple(config), self.loop
    ).result()

put(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig

Save a checkpoint to the database.

This method saves a checkpoint to the Postgres database. The checkpoint is associated with the provided config and its parent config (if any).

Parameters:

• config (RunnableConfig) –

  The config to associate with the checkpoint.

• checkpoint (Checkpoint) –

  The checkpoint to save.

• metadata (CheckpointMetadata) –

  Additional metadata to save with the checkpoint.

• new_versions (ChannelVersions) –

  New channel versions as of this write.

Returns:

• RunnableConfig ( RunnableConfig ) –

  Updated configuration after storing the checkpoint.

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Save a checkpoint to the database.

    This method saves a checkpoint to the Postgres database. The checkpoint is associated
    with the provided config and its parent config (if any).

    Args:
        config (RunnableConfig): The config to associate with the checkpoint.
        checkpoint (Checkpoint): The checkpoint to save.
        metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
        new_versions (ChannelVersions): New channel versions as of this write.

    Returns:
        RunnableConfig: Updated configuration after storing the checkpoint.
    """
    return asyncio.run_coroutine_threadsafe(
        self.aput(config, checkpoint, metadata, new_versions), self.loop
    ).result()

put_writes(config: RunnableConfig, writes: Sequence[tuple[str, Any]], task_id: str) -> None

Store intermediate writes linked to a checkpoint.

This method saves intermediate writes associated with a checkpoint to the database.

Parameters:

• config (RunnableConfig) –

  Configuration of the related checkpoint.

• writes (Sequence[Tuple[str, Any]]) –

  List of writes to store, each as (channel, value) pair.

• task_id (str) –

  Identifier for the task creating the writes.

Source code in libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

def put_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[tuple[str, Any]],
    task_id: str,
) -> None:
    """Store intermediate writes linked to a checkpoint.

    This method saves intermediate writes associated with a checkpoint to the database.

    Args:
        config (RunnableConfig): Configuration of the related checkpoint.
        writes (Sequence[Tuple[str, Any]]): List of writes to store, each as (channel, value) pair.
        task_id (str): Identifier for the task creating the writes.
    """
    return asyncio.run_coroutine_threadsafe(
        self.aput_writes(config, writes, task_id), self.loop
    ).result()

Comments