diff --git a/libervia/backend/memory/memory.py b/libervia/backend/memory/memory.py index 2666922b..fceb0271 100644 --- a/libervia/backend/memory/memory.py +++ b/libervia/backend/memory/memory.py @@ -17,18 +17,19 @@ # along with this program. If not, see . from collections import namedtuple +from collections.abc import Iterable import copy -from dataclasses import dataclass from functools import partial import mimetypes import os.path from pathlib import Path import time -from typing import Any, Dict, Optional, Tuple +from typing import Any, Dict, NamedTuple, Optional, Tuple from uuid import uuid4 import shortuuid from twisted.internet import defer, error, reactor +from twisted.internet.base import DelayedCall from twisted.python import failure from twisted.words.protocols.jabber import jid @@ -61,21 +62,39 @@ PresenceTuple = namedtuple("PresenceTuple", ("show", "priority", "statuses")) MSG_NO_SESSION = "Session id doesn't exist or is finished" -class Sessions(object): - """Sessions are data associated to key used for a temporary moment, with optional profile checking.""" +SessionData = dict[str, Any] + + +class SessionInfo(NamedTuple): + timer: DelayedCall + data: SessionData + profile: str | None = None + + +class Sessions: + """Temporary data handling. + + Sessions store data associated with a key for temporary use, with optional profile + validation. + """ DEFAULT_TIMEOUT = 600 - def __init__(self, timeout=None, resettable_timeout=True): + def __init__(self, timeout: int | None = None, resettable_timeout: bool = True): """ @param timeout (int): nb of seconds before session destruction @param resettable_timeout (bool): if True, the timeout is reset on each access """ - self._sessions = dict() + self._sessions: dict[str, SessionInfo] = dict() self.timeout = timeout or Sessions.DEFAULT_TIMEOUT self.resettable_timeout = resettable_timeout - def new_session(self, session_data=None, session_id=None, profile=None): + def new_session( + self, + session_data: SessionData | None = None, + session_id: str | None = None, + profile: str | None = None, + ) -> tuple[str, SessionData]: """Create a new session @param session_data: mutable data to use, default to a dict @@ -92,18 +111,12 @@ class Sessions(object): ) timer = reactor.callLater(self.timeout, self._purge_session, session_id) if session_data is None: - session_data = {} - self._sessions[session_id] = ( - (timer, session_data) if profile is None else (timer, session_data, profile) - ) + session_data = SessionData() + self._sessions[session_id] = SessionInfo(timer, session_data, profile) return session_id, session_data - def _purge_session(self, session_id): - try: - timer, session_data, profile = self._sessions[session_id] - except ValueError: - timer, session_data = self._sessions[session_id] - profile = None + def _purge_session(self, session_id: str): + timer, __, profile = self._sessions[session_id] try: timer.cancel() except error.AlreadyCalled: @@ -117,13 +130,13 @@ class Sessions(object): ) ) - def __len__(self): + def __len__(self) -> int: return len(self._sessions) - def __contains__(self, session_id): + def __contains__(self, session_id: str) -> bool: return session_id in self._sessions - def profile_get(self, session_id, profile): + def profile_get(self, session_id: str, profile: str) -> SessionData: try: timer, session_data, profile_set = self._sessions[session_id] except ValueError: @@ -138,30 +151,30 @@ class Sessions(object): timer.reset(self.timeout) return session_data - def __getitem__(self, session_id): + def __getitem__(self, session_id: str) -> SessionData: try: - timer, session_data = self._sessions[session_id] - except ValueError: - raise exceptions.InternalError( - "You need to use profile_get instead of __getitem__ when profile is set" - ) + session_info = self._sessions[session_id] except KeyError: raise failure.Failure(KeyError(MSG_NO_SESSION)) + if session_info.profile is not None: + raise exceptions.InternalError( + "You need to use profile_get instead of __getitem__ when profile is set." + ) if self.resettable_timeout: - timer.reset(self.timeout) - return session_data + session_info.timer.reset(self.timeout) + return session_info.data - def __setitem__(self, key, value): + def __setitem__(self, key: str, value) -> None: raise NotImplementedError("You need do use new_session to create a session") - def __delitem__(self, session_id): + def __delitem__(self, session_id: str) -> None: """delete the session data""" self._purge_session(session_id) - def keys(self): + def keys(self) -> list[str]: return list(self._sessions.keys()) - def iterkeys(self): + def iterkeys(self) -> Iterable[str]: return iter(self._sessions.keys()) @@ -198,7 +211,8 @@ class ProfileSessions(Sessions): ids = self._profile_get_all_ids(profile) if len(ids) > 1: raise exceptions.InternalError( - "profile_get_unique has been used but more than one session has been found!" + "profile_get_unique has been used but more than one session has been " + "found!" ) return ( self.profile_get(ids[0], profile) if len(ids) == 1 else None