Source code for langchain.memory.chat_message_histories.cassandra
"""Cassandra-based chat message history, based on cassIO."""
from __future__ import annotations
import json
import typing
from typing import List
if typing.TYPE_CHECKING:
    from cassandra.cluster import Session
from langchain.schema import (
    BaseChatMessageHistory,
)
from langchain.schema.messages import BaseMessage, _message_to_dict, messages_from_dict
DEFAULT_TABLE_NAME = "message_store"
DEFAULT_TTL_SECONDS = None
[docs]class CassandraChatMessageHistory(BaseChatMessageHistory):
    """Chat message history that stores history in Cassandra.
    Args:
        session_id: arbitrary key that is used to store the messages
            of a single chat session.
        session: a Cassandra `Session` object (an open DB connection)
        keyspace: name of the keyspace to use.
        table_name: name of the table to use.
        ttl_seconds: time-to-live (seconds) for automatic expiration
            of stored entries. None (default) for no expiration.
    """
[docs]    def __init__(
        self,
        session_id: str,
        session: Session,
        keyspace: str,
        table_name: str = DEFAULT_TABLE_NAME,
        ttl_seconds: int | None = DEFAULT_TTL_SECONDS,
    ) -> None:
        try:
            from cassio.history import StoredBlobHistory
        except (ImportError, ModuleNotFoundError):
            raise ImportError(
                "Could not import cassio python package. "
                "Please install it with `pip install cassio`."
            )
        self.session_id = session_id
        self.ttl_seconds = ttl_seconds
        self.blob_history = StoredBlobHistory(session, keyspace, table_name)
    @property