"""ChatSession model for RAG chatbot conversations.""" from datetime import datetime from typing import Optional from sqlalchemy import Column, String, DateTime, JSON, Text import uuid from src.config.database import Base class ChatSession(Base): """Represents a conversation session with the RAG chatbot.""" __tablename__ = "chat_sessions" id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4())) user_id = Column(String(255), nullable=True, index=True) conversation_history = Column(JSON, default=list) context = Column(JSON, default=dict) source_citations = Column(JSON, default=list) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) def to_dict(self) -> dict: """Convert model to dictionary.""" return { "id": str(self.id), "user_id": self.user_id, "conversation_history": self.conversation_history, "context": self.context, "source_citations": self.source_citations, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, } def add_message(self, role: str, content: str) -> None: """Add a message to conversation history. Args: role: Message role (user, assistant, system). content: Message content. """ if self.conversation_history is None: self.conversation_history = [] self.conversation_history.append({ "role": role, "content": content, "timestamp": datetime.utcnow().isoformat(), }) def add_citation(self, source: dict) -> None: """Add a source citation. Args: source: Source document information. """ if self.source_citations is None: self.source_citations = [] self.source_citations.append(source)