conversation_manager.py

# conversation_manager.py

import os
from supabase import create_client, Client
from typing import List, Dict, Any, Optional
import logging

logger = logging.getLogger(__name__)

class ConversationManager:
    def __init__(self):
        """Initializes the Supabase client."""
        supabase_url = os.getenv("SUPABASE_URL")
        supabase_key = os.getenv("SUPABASE_KEY")
        
        if not supabase_url or not supabase_key:
            raise ValueError("Supabase URL and Key must be set in environment variables.")
            
        self.supabase: Client = create_client(supabase_url, supabase_key)
        self.table_name = "conversations"
        logger.info("ConversationManager initialized with Supabase client.")

    def get_history(self, session_id: str) -> List[Dict[str, Any]]:
        """
        Retrieves conversation history for a given session_id.
        Returns an empty list if no history is found.
        """
        try:
            response = self.supabase.table(self.table_name)\
                .select("history")\
                .eq("session_id", session_id)\
                .limit(1)\
                .execute()
            
            if response.data:
                return response.data[0].get("history", [])
            return []
        except Exception as e:
            logger.error(f"Error fetching history for session {session_id}: {e}")
            return []

    def save_history(self, session_id: str, history: List[Dict[str, Any]]) -> None:
        """
        Saves or updates the conversation history for a session_id.
        Uses 'upsert' to create a new record or update an existing one.
        """
        try:
            self.supabase.table(self.table_name).upsert({
                "session_id": session_id,
                "history": history
            }).execute()
            logger.info(f"History for session {session_id} saved successfully.")
        except Exception as e:
            logger.error(f"Error saving history for session {session_id}: {e}")

    def delete_history(self, session_id: str) -> bool:
        """
        Deletes the conversation history for a given session_id.
        Returns True on success, False on failure.
        """
        try:
            self.supabase.table(self.table_name)\
                .delete()\
                .eq("session_id", session_id)\
                .execute()
            logger.info(f"History for session {session_id} deleted successfully.")
            return True
        except Exception as e:
            logger.error(f"Error deleting history for session {session_id}: {e}")
            return False