Source code for langchain_community.chat_message_histories.upstash_redis
import json
import logging
from typing import List, Optional
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 UpstashRedisChatMessageHistory(BaseChatMessageHistory):
    """Chat message history stored in an Upstash Redis database."""
[docs]    def __init__(
        self,
        session_id: str,
        url: str = "",
        token: str = "",
        key_prefix: str = "message_store:",
        ttl: Optional[int] = None,
    ):
        try:
            from upstash_redis import Redis
        except ImportError:
            raise ImportError(
                "Could not import upstash redis python package. "
                "Please install it with `pip install upstash_redis`."
            )
        if url == "" or token == "":
            raise ValueError(
                "UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN are needed."
            )
        try:
            self.redis_client = Redis(url=url, token=token)
        except Exception:
            logger.error("Upstash Redis instance could not be initiated.")
        self.session_id = session_id
        self.key_prefix = key_prefix
        self.ttl = ttl
    @property
    def key(self) -> str:
        """Construct the record key to use"""
        return self.key_prefix + self.session_id
    @property
    def messages(self) -> List[BaseMessage]:  # type: ignore
        """Retrieve the messages from Upstash Redis"""
        _items = self.redis_client.lrange(self.key, 0, -1)