import User from DBConn import DBConn, get_unix_timestamp class Session: """Class to hold a users login session methods and variables""" def __init__(self, session_ref, user: User.User or None): self.__session_id = None self.session_unique_ref = session_ref self.__user_id = None self.__session_ip = None self.__session_data = None self.__session_state = None self.__session_last_action_date = None self.__is_valid = False self.__user: User.User or None = user def __load_session(self) -> None: """Private method to load the users session from the database""" db_connection = DBConn() session_details = db_connection.do_select("sessions", "row", "*", "session_unique_ref = :session_unique_ref", [ { "field": "session_unique_ref", "value": self.session_unique_ref } ]) if session_details is not None and len(session_details) > 0: self.__session_id = session_details['session_id'] self.session_unique_ref = session_details['session_unique_ref'] self.__user_id = session_details['user_id'] self.__session_ip = session_details['session_ip'] self.__session_data = session_details['session_data'] self.__session_state = session_details['session_state'] self.__session_last_action_date = session_details['session_last_action_date'] self.__is_valid = True def session_is_valid(self): """Method to check if the session has a valid record loaded""" return self.__is_valid def load_session(self) -> None: """Method to load a session if the unique reference has been set""" if self.session_unique_ref: self.__load_session() def get_session_last_action_timestamp(self): """Method to get the sessions last action timestamp""" return self.__session_last_action_date def get_user(self) -> User.User or None: """Method to get the user object for the variable""" if self.__user_id: # Preload the user into the object if not loaded beforehand if not self.__user: self.__user = User.User(self.__user_id) return self.__user else: return None def update_session(self, current_state: str, last_action: str) -> None: """Method to update the session with the last action, current state and current timestamp""" self.__session_data = last_action self.__session_state = current_state self.__session_last_action_date = get_unix_timestamp() self.save_current_changes() def save_current_changes(self) -> bool: """Method to save the changes made to a session to the database""" result = False if self.__is_valid: db_connection = DBConn() if db_connection.do_update("sessions", [{"field": "session_data", "value": self.__session_data}, {"field": "session_state", "value": self.__session_state}, {"field": "session_last_action_date", "value": self.__session_last_action_date}], "session_id = :session_id", [{"field": "session_id", "value": self.__session_id}]): result = True return result def create_session(self, user_id: int, ip: str, data="", state="") -> bool: """Method to save a new session to the database""" result = False if not self.__is_valid: if user_id and len(ip) > 0: db_connection = DBConn() if db_connection.do_insert("sessions", [{"field": "session_unique_ref", "value": self.session_unique_ref}, {"field": "user_id", "value": user_id}, {"field": "session_ip", "value": ip}, {"field": "session_data", "value": data}, {"field": "session_state", "value": state}, {"field": "session_last_action_date", "value": get_unix_timestamp()}]): self.__load_session() result = True return result