haive.core.persistence.base

Base classes and interfaces for the Haive persistence system.

This module defines the core abstractions and interfaces for the persistence system used throughout the Haive framework. It provides the foundation for various persistence implementations, ensuring a consistent interface regardless of the underlying storage technology.

The central component is the CheckpointerConfig abstract base class, which defines the configuration interface that all persistence providers must implement. This allows different storage backends to be used interchangeably while providing a unified API for state persistence.

Classes

CheckpointerConfig

Base configuration for checkpoint persistence implementations.

Module Contents

class haive.core.persistence.base.CheckpointerConfig(/, **data)

Bases: pydantic.BaseModel, abc.ABC, Generic[T]

Base configuration for checkpoint persistence implementations.

This abstract base class defines the interface for all checkpointer configurations in the Haive framework. It provides a standardized way to configure and create checkpointer instances, regardless of the underlying storage technology.

The CheckpointerConfig class is generic over the connection type T, allowing different implementations to use appropriate connection objects (e.g., connection pools for databases, client objects for cloud services).

All concrete implementations must provide methods for creating both synchronous and asynchronous checkpointer instances, ensuring compatibility with both programming models.

Parameters:

data (Any)

type

The type of checkpointer (e.g., memory, postgres, sqlite)

mode

Operational mode - synchronous or asynchronous

storage_mode

Storage mode - full history or just the latest state

setup_needed

Whether tables/structures need to be created on first use

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

abstractmethod create_async_checkpointer()

Async:

Return type:

Any

Create an asynchronous checkpointer instance.

This method instantiates and configures an asynchronous checkpointer that matches the settings in this configuration. The returned object will be a compatible async checkpointer implementation that can be used for storing and retrieving state in asynchronous contexts.

Implementations should handle async connection management, setup of required database structures, and proper error handling with fallbacks.

Returns:

A configured asynchronous checkpointer instance

Return type:

Any

Raises:

RuntimeError – If the async checkpointer cannot be created due to missing dependencies or connection issues

Examples

config = PostgresCheckpointerConfig(mode=CheckpointerMode.ASYNC) async_checkpointer = await config.create_async_checkpointer() # Use with async graph graph = AsyncGraph(checkpointer=async_checkpointer)

abstractmethod create_checkpointer()

Create a synchronous checkpointer instance based on this configuration.

This method instantiates and configures a synchronous checkpointer that matches the settings in this configuration. The returned object will be a compatible checkpointer implementation (typically a LangGraph Saver) that can be used for storing and retrieving state.

Implementations should handle connection management, setup of required database structures, and proper error handling with fallbacks.

Returns:

A configured synchronous checkpointer instance

Return type:

Any

Raises:

RuntimeError – If the checkpointer cannot be created due to missing dependencies or connection issues

Examples

config = MemoryCheckpointerConfig() checkpointer = config.create_checkpointer() # Use checkpointer with a graph graph = Graph(checkpointer=checkpointer)

classmethod from_dict(data)

Create a checkpointer configuration from a dictionary.

This factory method deserializes a configuration from a dictionary, automatically selecting the appropriate implementation class based on the ‘type’ field. It provides a convenient way to create configurations from serialized data or configuration files.

The method dynamically imports the appropriate configuration class based on the specified type, ensuring minimal dependencies are loaded when not needed.

Parameters:

data (dict[str, Any]) – Dictionary containing configuration parameters, must include a ‘type’ field specifying which checkpointer implementation to use

Returns:

An instantiated checkpointer configuration of the

appropriate subclass

Return type:

CheckpointerConfig

Raises:

ValueError – If the ‘type’ field is missing or specifies an unsupported checkpointer type

Examples

config_dict = {

“type”: “postgres”, “db_host”: “localhost”, “db_port”: 5432

} config = CheckpointerConfig.from_dict(config_dict) # Returns a PostgresCheckpointerConfig instance

abstractmethod initialize_async_checkpointer()

Async:

Return type:

Any

Initialize an async checkpointer with proper resource management.

This method creates and initializes an asynchronous checkpointer with appropriate resource management. Unlike create_async_checkpointer, this method may return an async context manager that properly handles the lifecycle of resources like connection pools.

This is particularly important for database-backed checkpointers to ensure connections are properly closed when they’re no longer needed.

Returns:

An async context manager or the checkpointer itself, depending

on implementation requirements

Return type:

Any

Examples

config = PostgresCheckpointerConfig(mode=CheckpointerMode.ASYNC) async with await config.initialize_async_checkpointer() as checkpointer:

# Use checkpointer with async code # Resources will be properly closed after this block

is_async_mode()

Check if this configuration is set to operate in asynchronous mode.

This method determines whether the checkpointer should use asynchronous operations based on the configured mode. It’s used internally to choose between synchronous and asynchronous implementations.

Returns:

True if configured for async operations, False for synchronous

Return type:

bool

Examples

config = PostgresCheckpointerConfig(mode=CheckpointerMode.ASYNC) if config.is_async_mode():

# Use async methods checkpointer = await config.create_async_checkpointer()

else:

# Use sync methods checkpointer = config.create_checkpointer()

to_dict()

Convert this configuration to a dictionary.

This method serializes the configuration to a dictionary format, which is useful for persistence, logging, or passing to external systems. It automatically handles both Pydantic v1 and v2 models and excludes sensitive fields like passwords.

Returns:

Dictionary representation of this configuration

Return type:

Dict[str, Any]

Examples

config = PostgresCheckpointerConfig(db_host=”localhost”) config_dict = config.to_dict() print(config_dict) # {‘type’: ‘postgres’, ‘db_host’: ‘localhost’, …}