Spaces:
Build error
Build error
File size: 2,074 Bytes
ad8ba8a |
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 |
import uuid
import time
from typing import List, Dict, Optional
from src.domain.models import Session, SessionStatus, Message
from src.domain.interfaces import SessionRepository
class InMemorySessionRepository(SessionRepository):
def __init__(self):
self._sessions: Dict[str, Session] = {}
async def create_session(self) -> Session:
session_id = str(uuid.uuid4())
session = Session(
session_id=session_id,
title="New Chat",
status=SessionStatus.ACTIVE
)
self._sessions[session_id] = session
return session
async def get_session(self, session_id: str) -> Optional[Session]:
return self._sessions.get(session_id)
async def list_sessions(self) -> List[Session]:
return list(self._sessions.values())
async def delete_session(self, session_id: str) -> None:
if session_id in self._sessions:
del self._sessions[session_id]
async def update_session(self, session: Session) -> None:
self._sessions[session.session_id] = session
class SessionManager:
def __init__(self, repository: SessionRepository):
self.repository = repository
async def create_new_session(self) -> Session:
return await self.repository.create_session()
async def get_session_details(self, session_id: str) -> Optional[Session]:
return await self.repository.get_session(session_id)
async def list_all_sessions(self) -> List[Session]:
return await self.repository.list_sessions()
async def delete_session(self, session_id: str) -> None:
await self.repository.delete_session(session_id)
async def stop_session(self, session_id: str) -> None:
session = await self.repository.get_session(session_id)
if session:
session.status = SessionStatus.STOPPED
await self.repository.update_session(session)
# Singleton instance for simplicity in this demo
session_repository = InMemorySessionRepository()
session_manager = SessionManager(session_repository)
|