core (memory): Use better types for Sessions:

- `SessionData` is not a dedicated type instead of a simple `dict`.
- `SessionInfo` is now handled by a named tuple.
- Add type hints.

rel 465

libervia/backend/memory/memory.py

@@ -17,18 +17,19 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 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