Spaces:
Runtime error
Runtime error
File size: 4,653 Bytes
330b6e4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | """ChatSession model for managing user chat sessions."""
from datetime import datetime, timedelta
from sqlalchemy import Column, String, DateTime, Integer, Boolean, JSON
from sqlalchemy.orm import relationship
from .base import BaseModel, db, GUID
class ChatSession(BaseModel):
"""Model for managing user chat sessions."""
__tablename__ = 'chat_sessions'
# Core session fields
user_id = Column(GUID(), nullable=False, index=True)
language = Column(String(50), nullable=False, default='python')
# Activity tracking
last_active = Column(DateTime, default=datetime.utcnow, nullable=False, index=True)
message_count = Column(Integer, default=0, nullable=False)
is_active = Column(Boolean, default=True, nullable=False, index=True)
# Session metadata
session_metadata = Column(JSON, default=dict) # Additional session context, preferences, etc.
# Relationships
messages = relationship("Message", back_populates="session", cascade="all, delete-orphan")
language_context = relationship("LanguageContext", back_populates="session", uselist=False, cascade="all, delete-orphan")
def __init__(self, user_id, language='python', session_metadata=None):
"""Initialize a new chat session."""
super().__init__()
self.user_id = user_id
self.language = language
self.last_active = datetime.utcnow()
self.message_count = 0
self.is_active = True
self.session_metadata = session_metadata or {}
def update_activity(self):
"""Update the last active timestamp."""
self.last_active = datetime.utcnow()
db.session.commit()
def increment_message_count(self):
"""Increment the message count for this session."""
self.message_count += 1
self.update_activity()
def set_language(self, language):
"""Set the programming language for this session."""
self.language = language
self.update_activity()
def deactivate(self):
"""Mark the session as inactive."""
self.is_active = False
db.session.commit()
def is_expired(self, timeout_seconds=3600):
"""Check if the session has expired based on last activity."""
if not self.last_active:
return True
expiry_time = self.last_active + timedelta(seconds=timeout_seconds)
return datetime.utcnow() > expiry_time
def get_recent_messages(self, limit=10):
"""Get recent messages for this session."""
# Import here to avoid circular imports
from .message import Message
return (db.session.query(Message)
.filter(Message.session_id == self.id)
.order_by(Message.timestamp.desc())
.limit(limit)
.all())
def to_dict(self):
"""Convert session to dictionary with formatted timestamps."""
data = super().to_dict()
# Format timestamps as ISO strings for JSON serialization
if self.last_active:
data['last_active'] = self.last_active.isoformat()
return data
@classmethod
def create_session(cls, user_id, language='python', session_metadata=None):
"""Create a new chat session."""
session = cls(user_id=user_id, language=language, session_metadata=session_metadata)
db.session.add(session)
db.session.commit()
return session
@classmethod
def get_active_sessions(cls, user_id=None):
"""Get all active sessions, optionally filtered by user."""
query = db.session.query(cls).filter(cls.is_active == True)
if user_id:
query = query.filter(cls.user_id == user_id)
return query.all()
@classmethod
def cleanup_expired_sessions(cls, timeout_seconds=3600):
"""Clean up expired sessions."""
cutoff_time = datetime.utcnow() - timedelta(seconds=timeout_seconds)
expired_sessions = (db.session.query(cls)
.filter(cls.last_active < cutoff_time)
.filter(cls.is_active == True)
.all())
for session in expired_sessions:
session.deactivate()
return len(expired_sessions)
def __repr__(self):
"""String representation of the session."""
return f"<ChatSession(id={self.id}, user_id={self.user_id}, language={self.language}, active={self.is_active})>" |