Source code for langchain.memory.chat_message_histories.xata
import json
from typing import List
from langchain.schema import (
    BaseChatMessageHistory,
)
from langchain.schema.messages import BaseMessage, _message_to_dict, messages_from_dict
[docs]class XataChatMessageHistory(BaseChatMessageHistory):
    """Chat message history stored in a Xata database."""
[docs]    def __init__(
        self,
        session_id: str,
        db_url: str,
        api_key: str,
        branch_name: str = "main",
        table_name: str = "messages",
        create_table: bool = True,
    ) -> None:
        """Initialize with Xata client."""
        try:
            from xata.client import XataClient  # noqa: F401
        except ImportError:
            raise ValueError(
                "Could not import xata python package. "
                "Please install it with `pip install xata`."
            )
        self._client = XataClient(
            api_key=api_key, db_url=db_url, branch_name=branch_name
        )
        self._table_name = table_name
        self._session_id = session_id
        if create_table:
            self._create_table_if_not_exists()
    def _create_table_if_not_exists(self) -> None:
        r = self._client.table().get_schema(self._table_name)
        if r.status_code <= 299:
            return
        if r.status_code != 404:
            raise Exception(
                f"Error checking if table exists in Xata: {r.status_code} {r}"
            )
        r = self._client.table().create(self._table_name)
        if r.status_code > 299: