102 lines
4.7 KiB
Python
102 lines
4.7 KiB
Python
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
|