Created a FileSystemSessionStore class

plugins/filesystem_session/filesystem_session_store.py

+# -*- coding: utf-8 -*-
+
+import os
+import pickle
+import uuid
+
+from lodel.auth.exceptions import AuthenticationError
+from lodel.settings import Settings
+from lodel.utils.datetime import get_utc_timestamp
+
+
+class FileSystemSessionStore:
+
+    ## @brief instanciates a FileSystemSessionStore
+    # @param base_directory str : path to the base directory containing the session files (default: session.directory param of the settings)
+    # @param file_name_template str : template for the session files name (default : session.file_template param of the settings)
+    # @param expiration_limit int : duration of a session validity without any action made (defaut : session.expiration param of the settings)
+    def __init__(self, base_directory=Settings.session.directory, file_name_template=Settings.session.file_template, expiration_limit=Settings.session.expiration):
+        self.expiration_limit = expiration_limit
+        self.base_directory = base_directory
+        self.file_name_template = file_name_template
+
+    # === CRUD === #
+
+    ## @brief creates a new session and returns its id
+    # @param content dict : session content
+    # @return str
+    def create_new_session(self, content={}):
+        sid = FileSystemSessionStore.generate_new_sid()
+        session_file_path = self.get_session_file_path(sid)
+        session = content
+        self.save_session(sid, session)
+        return sid
+
+    ## @brief delete a session
+    # @param sid str : id of the session to be deleted
+    def delete_session(self, sid):
+        if self.is_session_existing(sid):
+            os.unlink(self.get_session_file_path(sid))
+        else:
+            raise AuthenticationError("No session file found for the sid %s" % sid)
+
+    ## @brief gets a session's content
+    # @param sid str : id of the session to read
+    def get_session(self, sid):
+        if self.is_session_existing(sid):
+            if not self.has_session_expired(sid):
+                session = self.get_session(sid)
+            else:
+                self.delete_session(sid)
+                session = {}
+        else:
+            raise AuthenticationError("No session file found for the sid %s" % sid)
+
+        return session
+
+    ## @brief updates a session's content
+    # @param sid str : session's id
+    # @param content dict : items to update with their new value
+    def update_session(self, sid, content):
+        session = self.get_session(sid)
+        for key, value in content.items():
+            if key != 'sid':
+                session[key] = value
+        self.save_session(sid, session)
+
+    ## @brief lists all the session files paths
+    # @return list
+    def list_all_sessions(self):
+        session_files_directory = os.path.abspath(self.base_directory)
+        session_files_list = [os.path.join(session_files_directory, file_object) for file_object in os.listdir(session_files_directory) if os.path.isfile(os.path.join(session_files_directory, file_object))]
+        return session_files_list
+
+    def clean(self):
+        files_list = self.list_all_sessions()
+        now_timestamp = get_utc_timestamp()
+        for session_file in files_list:
+            if self.has_session_file_expired(session_file):
+                os.unlink(session_file)
+
+    # === UTILS === #
+    @classmethod
+    def generate_new_sid(cls):
+        return uuid.uuid1()
+
+    ## @brief returns the file path for a given session id
+    # @param sid str : session id
+    # @return str
+    def get_session_file_path(self, sid):
+        return os.path.join(self.base_directory, self.file_name_template) % sid
+
+    ## @brief saves a session to a file
+    # @param sid str : session id
+    # @param session dict : content to be saved
+    def save_session(self, sid, session):
+        session_file_path = self.get_session_file_path(sid)
+        pickle.dump(session, open(session_file_path, "wb"))
+
+    ## @brief checks if a session exists
+    # @param sid str : session id
+    # @return bool
+    def is_session_existing(self, sid):
+        session_file = self.get_session_file_path(sid)
+        return os.path.is_file(session_file)
+
+    ## @brief checks if a session has expired
+    # @param sid str: session id
+    # @return bool
+    def has_session_expired(self, sid):
+        session_file = self.get_session_file_path(sid)
+        return self.has_session_file_expired(session_file)
+
+    def has_session_file_expired(self, session_file):
+        expiration_timestamp = os.stat(session_file).st_mtime + self.expiration_limit
+        now_timestamp = get_utc_timestamp()
+        return now_timestamp > expiration_timestamp

plugins/filesystem_session/main.py

 # -*- coding: utf-8 -*-
 
-import os
-import pickle
-import uuid
-
 from lodel.auth.exceptions import AuthenticationError
 from lodel.plugin import LodelHook
-from lodel.settings import Settings
-from lodel.utils.datetime import get_utc_timestamp
-
-SESSION_FILES_BASE_DIR = Settings.sessions.directory
-SESSION_FILES_TEMPLATE = Settings.sessions.file_template
-SESSION_EXPIRATION_LIMIT = Settings.sessions.expiration
-
-
-## @brief generates a new session id
-def generate_new_sid():
-    return uuid.uuid1()
-
-
-## @brief gets the session file path, given a session id
-# @param sid str : session id
-# @return str
-def get_session_file_path(sid):
-    return os.path.join(SESSION_FILES_BASE_DIR, SESSION_FILES_TEMPLATE) % sid
-
-
-## @brief saves the session content to a session file
-# @param sid str: session id
-# @param session dict : session content to be saved
-def save_session(sid, session):
-    session_file_path = get_session_file_path(sid)
-    pickle.dump(session, open(session_file_path, "wb"))
 
+from .filesystem_session_store import FileSystemSessionStore
 
-## @brief checks if a session file has expired
-# @param sid str : session id
-# @return bool
-def is_session_expired(sid):
-    session_file = get_session_file_path(sid)
-    if not os.path.isfile(session_file):
-        raise AuthenticationError("No session file found for the sid : %s" % sid)
-
-    expiration_timestamp = os.stat(session_file).st_mtime + SESSION_EXPIRATION_LIMIT
-    now_timestamp = get_utc_timestamp()
-    return now_timestamp >= expiration_timestamp
-
-
-## @brief reads a session content
-# @param sid str: session id
-# @return dict
-def get_session_content(sid):
-    session_file_path = get_session_file_path(sid)
-    return pickle.load(open(session_file_path), 'rb')
+session_store = FileSystemSessionStore()
 
 
 ## @brief starts a new session and returns its sid
 # @return str
 @LodelHook('session_start')
 def start_session(caller, payload):
-    sid = generate_new_sid()
-    session_file = get_session_file_path(sid)
-    session = dict()
-    for key, value in payload.items():
-        session[key] = value
-    save_session(sid, session)
-    return sid
+    return session_store.create_new_session(payload)
 
-## @brief stops a session
+
+## @brief destroys a session
 # @param caller *
 # @param sid str : session id
 @LodelHook('session_destroy')
 def stop_session(caller, sid):
-    session_file_path = get_session_file_path(sid)
-    if os.path.isfile(session_file_path):
-        os.unlink(session_file_path)
-    else:
-        raise AuthenticationError("No session file found for the sid %s" % sid)
+    session_store.delete_session(sid)
 
 
 ## @brief reads a session content
 # @param caller *
 # @param sid str: session id
+# @return dict
 @LodelHook('session_load')
 def read_session(caller, sid):
-    session_file = get_session_file_path(sid)
-    if os.path.isfile(session_file):
-        if not is_session_expired(sid):
-            session = get_session_content(sid)
-        else:
-            LodelHook.call_hook('session_stop', __file__, sid)
-            session = {}
-    else:
-        raise AuthenticationError("No session file found for the sid %s" % sid)
-    return session
+    return session_store.get_session(sid)
 
 
-## @brief deletes all old session files (expired ones)
+## @brief destroys all the old sessions (expired ones)
 # @param caller *
 @LodelHook('session_clean')
 def clean_sessions(caller):
-    session_files_path = os.path.abspath(SESSION_FILES_BASE_DIR)
-    session_files = [os.path.join(session_files_path, file_object) for file_object in os.listdir(session_files_path) if os.path.isfile(os.path.join(session_files_path, file_object))]
-    now_timestamp = get_utc_timestamp()
-    for session_file in session_files:
-        last_modified = os.stat(session_file).st_mtime
-        expiration_timestamp = last_modified + SESSION_EXPIRATION_LIMIT
-        if now_timestamp > expiration_timestamp:
-            os.unlink(session_file)
+    session_store.clean()
+
 
-## @brief updates the content session
+## @brief updates the content of the session
 # @param caller *
-# @param payload dict: datas to insert/update in the session
+# @param payload dict : datas to insert/update in the session
 @LodelHook('update_session')
 def update_session_content(caller, payload):
     if 'sid' in payload:
         sid = payload['sid']
-        session = LodelHook.call_hook('session_load', __file__, sid)
-        for key, value in payload.items():
-            session[key] = value
-        save_session(sid, session)
+        del payload['sid']
+        session_store.update_session(sid, payload)
     else:
-        raise AuthenticationError("Missing session id in the request")
+        raise AuthenticationError("Session Update: Missing sid (session id) in the given payload argument")