Source code for langchain_community.chat_message_histories.singlestoredb
import json
import logging
import re
from typing import (
    Any,
    List,
)
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import (
    BaseMessage,
    message_to_dict,
    messages_from_dict,
)
logger = logging.getLogger(__name__)
[docs]class SingleStoreDBChatMessageHistory(BaseChatMessageHistory):
    """Chat message history stored in a SingleStoreDB database."""
[docs]    def __init__(
        self,
        session_id: str,
        *,
        table_name: str = "message_store",
        id_field: str = "id",
        session_id_field: str = "session_id",
        message_field: str = "message",
        pool_size: int = 5,
        max_overflow: int = 10,
        timeout: float = 30,
        **kwargs: Any,
    ):
        """Initialize with necessary components.
        Args:
            table_name (str, optional): Specifies the name of the table in use.
                Defaults to "message_store".
            id_field (str, optional): Specifies the name of the id field in the table.
                Defaults to "id".
            session_id_field (str, optional): Specifies the name of the session_id
                field in the table. Defaults to "session_id".
            message_field (str, optional): Specifies the name of the message field
                in the table. Defaults to "message".
            Following arguments pertain to the connection pool:
            pool_size (int, optional): Determines the number of active connections in
                the pool. Defaults to 5.
            max_overflow (int, optional): Determines the maximum number of connections