from abc import ABC, abstractmethod
from typing import Optional, Dict
from .session import Session
class SessionRepository(ABC):
@abstractmethod
async def save_session(self, session: Session):
"""Persists session data (DB, Redis, etc.)."""
pass
@abstractmethod
async def get_session(self, session_id: str) -> Optional[Session]:
"""Retrieves session data."""
pass
@abstractmethod
async def delete_session(self, session_id: str):
"""Removes session data."""
pass
[docs]
class InMemorySessionRepository(SessionRepository):
def __init__(self):
self.sessions: Dict[str, Session] = {}
[docs]
async def save_session(self, session: Session):
"""Stores session in-memory."""
self.sessions[session.session_id] = session
[docs]
async def get_session(self, session_id: str) -> Session:
"""Retrieves session from memory, or creates a new one if not found."""
session = self.sessions.get(session_id)
if session is None:
session = Session(session_id=session_id)
self.sessions[session_id] = session
return session
[docs]
async def delete_session(self, session_id: str):
"""Deletes session from memory."""
self.sessions.pop(session_id, None)