Spaces:
Runtime error
Runtime error
File size: 4,355 Bytes
f3997d4 | 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 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 | from sqlalchemy.orm import Session
from sqlalchemy import func
from typing import List, Optional
from datetime import datetime
from app.database.models import Session as ChatSession, Message
from app.utils.helpers import generate_id
class SessionService:
"""Service for session management operations."""
@staticmethod
def create_session(
db: Session,
user_id: Optional[str] = None,
title: str = "New Conversation"
) -> ChatSession:
"""
Create a new chat session.
Args:
db: Database session
user_id: Optional user ID (None for guest users)
title: Session title
Returns:
Created ChatSession object
"""
session = ChatSession(
id=generate_id(),
user_id=user_id,
title=title
)
db.add(session)
db.commit()
db.refresh(session)
return session
@staticmethod
def get_user_sessions(db: Session, user_id: str) -> List[dict]:
"""
Get all sessions for a user with message counts.
Args:
db: Database session
user_id: User ID
Returns:
List of session dictionaries with message counts
"""
# Query sessions with message counts
sessions = db.query(
ChatSession,
func.count(Message.id).label('message_count')
).outerjoin(
Message, ChatSession.id == Message.session_id
).filter(
ChatSession.user_id == user_id
).group_by(
ChatSession.id
).order_by(
ChatSession.updated_at.desc()
).all()
# Format response
result = []
for session, message_count in sessions:
result.append({
"id": session.id,
"user_id": session.user_id,
"title": session.title,
"summary": session.summary,
"created_at": session.created_at,
"updated_at": session.updated_at,
"message_count": message_count
})
return result
@staticmethod
def get_session_by_id(db: Session, session_id: str) -> Optional[ChatSession]:
"""
Get session by ID.
Args:
db: Database session
session_id: Session ID
Returns:
ChatSession object if found, None otherwise
"""
return db.query(ChatSession).filter(ChatSession.id == session_id).first()
@staticmethod
def update_session_title(db: Session, session_id: str, title: str) -> Optional[ChatSession]:
"""
Update session title.
Args:
db: Database session
session_id: Session ID
title: New title
Returns:
Updated ChatSession object if found, None otherwise
"""
session = db.query(ChatSession).filter(ChatSession.id == session_id).first()
if not session:
return None
session.title = title
session.updated_at = datetime.utcnow()
db.commit()
db.refresh(session)
return session
@staticmethod
def delete_session(db: Session, session_id: str) -> bool:
"""
Delete a session and all its messages.
Args:
db: Database session
session_id: Session ID
Returns:
True if deleted, False if not found
"""
session = db.query(ChatSession).filter(ChatSession.id == session_id).first()
if not session:
return False
db.delete(session)
db.commit()
return True
@staticmethod
def update_session_timestamp(db: Session, session_id: str):
"""
Update session's updated_at timestamp.
Args:
db: Database session
session_id: Session ID
"""
session = db.query(ChatSession).filter(ChatSession.id == session_id).first()
if session:
session.updated_at = datetime.utcnow()
db.commit()
|