"""Session model and in-memory session store."""
import uuid
from abc import ABC, abstractmethod
from datetime import datetime
from pydantic import BaseModel, Field
from navi.llm.base import Message
class Session(BaseModel):
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
profile_id: str
messages: list[Message] = Field(default_factory=list)
created_at: datetime = Field(default_factory=datetime.utcnow)
last_active: datetime = Field(default_factory=datetime.utcnow)
class SessionStore(ABC):
@abstractmethod
async def create(self, profile_id: str) -> Session: ...
@abstractmethod
async def get(self, session_id: str) -> Session | None: ...
@abstractmethod
async def save(self, session: Session) -> None: ...
@abstractmethod
async def list_all(self) -> list[Session]: ...
@abstractmethod
async def delete(self, session_id: str) -> bool: ...
class InMemorySessionStore(SessionStore):
def __init__(self) -> None:
self._sessions: dict[str, Session] = {}
async def create(self, profile_id: str) -> Session:
session = Session(profile_id=profile_id)
self._sessions[session.id] = session
return session
async def get(self, session_id: str) -> Session | None:
return self._sessions.get(session_id)
async def save(self, session: Session) -> None:
session.last_active = datetime.utcnow()
self._sessions[session.id] = session
async def list_all(self) -> list[Session]:
return list(self._sessions.values())
async def delete(self, session_id: str) -> bool:
if session_id in self._sessions:
del self._sessions[session_id]
return True
return False