haive.core.persistence.base
===========================

.. py:module:: haive.core.persistence.base

.. autoapi-nested-parse::

   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
-------

.. autoapisummary::

   haive.core.persistence.base.CheckpointerConfig


Module Contents
---------------

.. py:class:: CheckpointerConfig(/, **data)

   Bases: :py:obj:`pydantic.BaseModel`, :py:obj:`abc.ABC`, :py:obj:`Generic`\ [\ :py:obj:`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.

   .. attribute:: type

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

   .. attribute:: mode

      Operational mode - synchronous or asynchronous

   .. attribute:: storage_mode

      Storage mode - full history or just the latest state

   .. attribute:: 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.


   .. py:method:: create_async_checkpointer()
      :abstractmethod:

      :async:


      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
      :rtype: Any

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

      .. rubric:: Examples

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



   .. py:method:: create_checkpointer()
      :abstractmethod:


      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
      :rtype: Any

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

      .. rubric:: Examples

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



   .. py:method:: from_dict(data)
      :classmethod:


      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.

      :param data: Dictionary containing configuration parameters, must include a
                   'type' field specifying which checkpointer implementation to use

      :returns:

                An instantiated checkpointer configuration of the
                    appropriate subclass
      :rtype: CheckpointerConfig

      :raises ValueError: If the 'type' field is missing or specifies an unsupported
          checkpointer type

      .. rubric:: Examples

      config_dict = {
          "type": "postgres",
          "db_host": "localhost",
          "db_port": 5432
      }
      config = CheckpointerConfig.from_dict(config_dict)
      # Returns a PostgresCheckpointerConfig instance



   .. py:method:: initialize_async_checkpointer()
      :abstractmethod:

      :async:


      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
      :rtype: Any

      .. rubric:: 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



   .. py:method:: 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
      :rtype: bool

      .. rubric:: 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()



   .. py:method:: 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
      :rtype: Dict[str, Any]

      .. rubric:: Examples

      config = PostgresCheckpointerConfig(db_host="localhost")
      config_dict = config.to_dict()
      print(config_dict)  # {'type': 'postgres', 'db_host': 'localhost', ...}