[MIG] session_redis: Migration to 18.0

session_redis/__manifest__.py

@@ -14,6 +14,5 @@
         "python": ["redis"],
     },
     "website": "https://github.com/camptocamp/odoo-cloud-platform",
-    "data": [],
-    "installable": False,
+    "installable": True,
 }

session_redis/http.py

@@ -3,6 +3,8 @@
 
 import logging
 import os
+from pathlib import Path
+from typing import Optional
 
 from odoo import http
 from odoo.tools import config
@@ -21,29 +23,32 @@ except ImportError:
     _logger.debug("Cannot 'import redis'.")
 
 
-def is_true(strval):
-    return bool(strtobool(strval or "0".lower()))
+def is_true(strval: Optional[str]) -> bool:
+    """Convert string value to boolean."""
+    return bool(strtobool((strval or "0").lower()))
 
 
-sentinel_host = os.environ.get("ODOO_SESSION_REDIS_SENTINEL_HOST")
-sentinel_master_name = os.environ.get("ODOO_SESSION_REDIS_SENTINEL_MASTER_NAME")
+# Retrieve Redis session configurations from environment variables
+sentinel_host = os.getenv("ODOO_SESSION_REDIS_SENTINEL_HOST")
+sentinel_master_name = os.getenv("ODOO_SESSION_REDIS_SENTINEL_MASTER_NAME")
 if sentinel_host and not sentinel_master_name:
     raise Exception(
         "ODOO_SESSION_REDIS_SENTINEL_MASTER_NAME must be defined "
         "when using session_redis"
     )
-sentinel_port = int(os.environ.get("ODOO_SESSION_REDIS_SENTINEL_PORT", 26379))
-host = os.environ.get("ODOO_SESSION_REDIS_HOST", "localhost")
-port = int(os.environ.get("ODOO_SESSION_REDIS_PORT", 6379))
-prefix = os.environ.get("ODOO_SESSION_REDIS_PREFIX")
-url = os.environ.get("ODOO_SESSION_REDIS_URL")
-password = os.environ.get("ODOO_SESSION_REDIS_PASSWORD")
-expiration = os.environ.get("ODOO_SESSION_REDIS_EXPIRATION")
-anon_expiration = os.environ.get("ODOO_SESSION_REDIS_EXPIRATION_ANONYMOUS")
+sentinel_port = int(os.getenv("ODOO_SESSION_REDIS_SENTINEL_PORT", 26379))
+host = os.getenv("ODOO_SESSION_REDIS_HOST", "localhost")
+port = int(os.getenv("ODOO_SESSION_REDIS_PORT", 6379))
+prefix = os.getenv("ODOO_SESSION_REDIS_PREFIX")
+url = os.getenv("ODOO_SESSION_REDIS_URL")
+password = os.getenv("ODOO_SESSION_REDIS_PASSWORD")
+expiration = os.getenv("ODOO_SESSION_REDIS_EXPIRATION")
+anon_expiration = os.getenv("ODOO_SESSION_REDIS_EXPIRATION_ANONYMOUS")
 
 
 @lazy_property
-def session_store(self):
+def session_store(self) -> RedisSessionStore:
+    """Configure Redis session storage."""
     if sentinel_host:
         sentinel = Sentinel([(sentinel_host, sentinel_port)], password=password)
         redis_client = sentinel.master_for(sentinel_master_name)
@@ -61,30 +66,32 @@ def session_store(self):
 
 
 def purge_fs_sessions(path):
-    for fname in os.listdir(path):
-        path = os.path.join(path, fname)
+    """Remove old file-based sessions."""
+    session_path = Path(path)
+    if not session_path.exists():
+        _logger.warning(f"Session directory '{session_path}' does not exist.")
+        return
+
+    for session_file in session_path.iterdir():
         try:
-            os.unlink(path)
-        except OSError:
-            _logger.warning("OS Error during purge of redis sessions.")
+            session_file.unlink()
+            _logger.debug(f"Deleted session file: {session_file}")
+        except PermissionError:
+            _logger.warning(f"Permission denied while deleting session file: {session_file}")
+        except OSError as e:
+            _logger.warning(f"Error deleting session file {session_file}: {str(e)}")
 
 
-if is_true(os.environ.get("ODOO_SESSION_REDIS")):
+if is_true(os.getenv("ODOO_SESSION_REDIS")):
+    storage_info = f"Redis with prefix '{prefix}' on "
     if sentinel_host:
-        _logger.debug(
-            "HTTP sessions stored in Redis with prefix '%s'. "
-            "Using Sentinel on %s:%s",
-            prefix or "",
-            sentinel_host,
-            sentinel_port,
-        )
+        storage_info += f"Sentinel {sentinel_host}:{sentinel_port}"
     else:
-        _logger.debug(
-            "HTTP sessions stored in Redis with prefix '%s' on " "%s:%s",
-            prefix or "",
-            host,
-            port,
-        )
+        storage_info += f"{host}:{port}"
+
+    _logger.debug("HTTP sessions stored in %s.", storage_info)
+
     http.Application.session_store = session_store
-    # clean the existing sessions on the file system
+
+    # Clean existing sessions stored in the file system
     purge_fs_sessions(config.session_dir)

session_redis/json_encoding.py

@@ -3,8 +3,9 @@
 
 import json
 from datetime import date, datetime
+from typing import Any, Dict
 
-import dateutil
+import dateutil.parser
 
 
 class SessionEncoder(json.JSONEncoder):
@@ -13,30 +14,30 @@ class SessionEncoder(json.JSONEncoder):
     So that we can later recompose them if they were stored in the session
     """
 
-    def default(self, obj):
+    def default(self, obj: Any) -> Any:
         if isinstance(obj, datetime):
             return {"_type": "datetime_isoformat", "value": obj.isoformat()}
-        elif isinstance(obj, date):
+        if isinstance(obj, date):
             return {"_type": "date_isoformat", "value": obj.isoformat()}
-        elif isinstance(obj, set):
-            return {"_type": "set", "value": tuple(obj)}
-        return json.JSONEncoder.default(self, obj)
+        if isinstance(obj, set):
+            return {"_type": "set", "value": tuple(sorted(obj))}
+        return super().default(obj)
 
 
 class SessionDecoder(json.JSONDecoder):
     """Decode json, recomposing recordsets and date/datetime"""
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, object_hook=self.object_hook, **kwargs)
 
-    def object_hook(self, obj):
+    def object_hook(self, obj: Dict[str, Any]) -> Any:
+        """Convert serialized data back into its original Python type."""
         if "_type" not in obj:
             return obj
-        type_ = obj["_type"]
-        if type_ == "datetime_isoformat":
+        if obj["_type"] == "datetime_isoformat":
             return dateutil.parser.parse(obj["value"])
-        elif type_ == "date_isoformat":
+        if obj["_type"] == "date_isoformat":
             return dateutil.parser.parse(obj["value"]).date()
-        elif type_ == "set":
+        if obj["_type"] == "set":
             return set(obj["value"])
         return obj

session_redis/session.py

@@ -3,6 +3,7 @@
 
 import json
 import logging
+from typing import Optional, List, Dict, Any
 
 from odoo.service import security
 from odoo.tools._vendor.sessions import SessionStore
@@ -24,93 +25,72 @@ class RedisSessionStore(SessionStore):
         self,
         redis,
         session_class=None,
-        prefix="",
-        expiration=None,
-        anon_expiration=None,
-    ):
+        prefix: str = "",
+        expiration: Optional[int] = None,
+        anon_expiration: Optional[int] = None,
+    ) -> None:
         super().__init__(session_class=session_class)
         self.redis = redis
-        if expiration is None:
-            self.expiration = DEFAULT_SESSION_TIMEOUT
-        else:
-            self.expiration = expiration
-        if anon_expiration is None:
-            self.anon_expiration = DEFAULT_SESSION_TIMEOUT_ANONYMOUS
-        else:
-            self.anon_expiration = anon_expiration
-        self.prefix = "session:"
-        if prefix:
-            self.prefix = f"{self.prefix}:{prefix}:"
+        self.expiration = expiration if expiration is not None else DEFAULT_SESSION_TIMEOUT
+        self.anon_expiration = anon_expiration if anon_expiration is not None else DEFAULT_SESSION_TIMEOUT_ANONYMOUS
+        self.prefix = f"session:{prefix}:" if prefix else "session:"
 
-    def build_key(self, sid):
+    def build_key(self, sid: str) -> str:
+        """Build the Redis key for a session ID."""
         return f"{self.prefix}{sid}"
 
-    def save(self, session):
+    def save(self, session) -> Optional[bool]:
+        """Save session data in Redis with an expiration time."""
         key = self.build_key(session.sid)
 
         # allow to set a custom expiration for a session
         # such as a very short one for monitoring requests
-        if session.uid:
-            expiration = session.expiration or self.expiration
-        else:
-            expiration = session.expiration or self.anon_expiration
-        if _logger.isEnabledFor(logging.DEBUG):
-            if session.uid:
-                user_msg = f"user '{session.login}' (id: {session.uid})"
-            else:
-                user_msg = "anonymous user"
-            _logger.debug(
-                "saving session with key '%s' and " "expiration of %s seconds for %s",
-                key,
-                expiration,
-                user_msg,
-            )
+        expiration = session.expiration or (self.expiration if session.uid else self.anon_expiration)
 
-        data = json.dumps(dict(session), cls=json_encoding.SessionEncoder).encode(
-            "utf-8"
-        )
+        if _logger.isEnabledFor(logging.DEBUG):
+            user_msg = f"user '{session.login}' (id: {session.uid})" if session.uid else "anonymous user"
+            _logger.debug(f"Saving session '{key}' with expiration {expiration} seconds for {user_msg}")
+
+        data = json.dumps(dict(session), cls=json_encoding.SessionEncoder).encode("utf-8")
         if self.redis.set(key, data):
             return self.redis.expire(key, expiration)
+        return None
 
-    def delete(self, session):
+    def delete(self, session) -> int:
+        """Delete a session from Redis."""
         key = self.build_key(session.sid)
-        _logger.debug("deleting session with key %s", key)
+        _logger.debug(f"Deleting session with '{key}'")
         return self.redis.delete(key)
 
-    def get(self, sid):
+    def get(self, sid: str):
+        """Retrieve a session from Redis, or return a new one if not found."""
         if not self.is_valid_key(sid):
-            _logger.debug(
-                "session with invalid sid '%s' has been asked, " "returning a new one",
-                sid,
-            )
+            _logger.debug(f"Invalid session ID '{sid}' requested, returning a new session.")
             return self.new()
 
         key = self.build_key(sid)
         saved = self.redis.get(key)
         if not saved:
-            _logger.debug(
-                "session with non-existent key '%s' has been asked, "
-                "returning a new one",
-                key,
-            )
+            _logger.debug(f"Non-existent session '{key}' requested, returning a new session.")
             return self.new()
+
         try:
             data = json.loads(saved.decode("utf-8"), cls=json_encoding.SessionDecoder)
-        except ValueError:
-            _logger.debug(
-                "session for key '%s' has been asked but its json "
-                "content could not be read, it has been reset",
-                key,
-            )
+        except (ValueError, json.JSONDecodeError):
+            _logger.warning(f"Corrupt session data for key '{key}', resetting session.")
             data = {}
+
         return self.session_class(data, sid, False)
 
-    def list(self):
-        keys = self.redis.keys("%s*" % self.prefix)
+    def list(self) -> List[str]:
+        """List all session keys in Redis."""
+        # More efficient scanning
+        keys = [key for key in self.redis.scan_iter(f"{self.prefix}*")]
         _logger.debug("a listing redis keys has been called")
-        return [key[len(self.prefix) :] for key in keys]
+        return [key.decode("utf-8")[len(self.prefix) :] for key in keys]
 
-    def rotate(self, session, env):
+    def rotate(self, session, env) -> None:
+        """Rotate session ID and regenerate session token if user is logged in."""
         self.delete(session)
         session.sid = self.generate_key()
         if session.uid and env:

session_redis/strtobool.py

@@ -1,3 +1,5 @@
+from typing import Union
+
 _MAP = {
     "y": True,
     "yes": True,
@@ -14,8 +16,11 @@ _MAP = {
 }
 
 
-def strtobool(value):
-    try:
-        return _MAP[str(value).lower()]
-    except KeyError as error:
-        raise ValueError(f'"{value}" is not a valid bool value') from error
+def strtobool(value: Union[str, int]) -> bool:
+    """Convert a string or integer representation to a boolean value."""
+    result = _MAP.get(str(value).strip().lower())
+
+    if result is None:
+        raise ValueError(f"Invalid boolean value: {repr(value)}")
+
+    return result