28 lines
1,007 B
Python
28 lines
1,007 B
Python
from datetime import datetime
|
|
|
|
from domain.sandbox import SandboxSession
|
|
from usecase.interface import SandboxSessionRepository
|
|
|
|
|
|
class InMemorySandboxSessionRepository(SandboxSessionRepository):
|
|
def __init__(self) -> None:
|
|
self._sessions_by_chat_id: dict[str, SandboxSession] = {}
|
|
|
|
def get_active_by_chat_id(self, chat_id: str) -> SandboxSession | None:
|
|
return self._sessions_by_chat_id.get(chat_id)
|
|
|
|
def list_expired(self, now: datetime) -> list[SandboxSession]:
|
|
return [
|
|
session
|
|
for session in self._sessions_by_chat_id.values()
|
|
if session.expires_at <= now
|
|
]
|
|
|
|
def save(self, session: SandboxSession) -> None:
|
|
self._sessions_by_chat_id[session.chat_id] = session
|
|
|
|
def delete(self, session_id: str) -> None:
|
|
for chat_id, session in tuple(self._sessions_by_chat_id.items()):
|
|
if session.session_id == session_id:
|
|
del self._sessions_by_chat_id[chat_id]
|
|
return
|