Source code for langchain.memory.chat_message_histories.cassandra
import json
import logging
from typing import List
from langchain.schema import (
    BaseChatMessageHistory,
    BaseMessage,
    _message_to_dict,
    messages_from_dict,
)
logger = logging.getLogger(__name__)
DEFAULT_KEYSPACE_NAME = "chat_history"
DEFAULT_TABLE_NAME = "message_store"
DEFAULT_USERNAME = "cassandra"
DEFAULT_PASSWORD = "cassandra"
DEFAULT_PORT = 9042
[docs]class CassandraChatMessageHistory(BaseChatMessageHistory):
    """Chat message history that stores history in Cassandra.
    Args:
        contact_points: list of ips to connect to Cassandra cluster
        session_id: arbitrary key that is used to store the messages
            of a single chat session.
        port: port to connect to Cassandra cluster
        username: username to connect to Cassandra cluster
        password: password to connect to Cassandra cluster
        keyspace_name: name of the keyspace to use
        table_name: name of the table to use
    """
    def __init__(
        self,
        contact_points: List[str],
        session_id: str,
        port: int = DEFAULT_PORT,
        username: str = DEFAULT_USERNAME,
        password: str = DEFAULT_PASSWORD,
        keyspace_name: str = DEFAULT_KEYSPACE_NAME,
        table_name: str = DEFAULT_TABLE_NAME,
    ):
        self.contact_points = contact_points
        self.session_id = session_id
        self.port = port
        self.username = username
        self.password = password
        self.keyspace_name = keyspace_name
        self.table_name = table_name
        try:
            from cassandra import (
                AuthenticationFailed,
                OperationTimedOut,
                UnresolvableContactPoints,
            )
rtdocs_stable/api.python.langchain.com/en/stable/_modules/langchain/memory/chat_message_histories/cassandra.html