ultimate_tictactoe_server/session_manager.py

60 lines
2.8 KiB
Python
Raw Normal View History

2019-03-14 00:01:02 +01:00
from database_connection import DatabaseConnection, SQLInjectionError, get_sql_time
from user import User
import datetime
import uuid
class SessionManager(object):
def __init__(self, session_lifespan_timedelta):
self.session_lifespan_timedelta = session_lifespan_timedelta
def get_session_by_id(self, session_id):
query = f"SELECT * FROM sessions WHERE id='{session_id}'"
return DatabaseConnection.global_single_query(query)
def touch_session(self, session_id):
query = f"UPDATE sessions SET last_seen='{datetime.datetime.now()}' WHERE id='{session_id}'"
DatabaseConnection.global_single_execution(query)
def get_session_by_registered_user(self, user_name):
query = f"SELECT * FROM sessions WHERE registered_user='{user_name}'"
return DatabaseConnection.global_single_query(query)
def get_session_by_temp_user(self, user_name):
query = f"SELECT * FROM sessions WHERE temp_user='{user_name}'"
return DatabaseConnection.global_single_query(query)
def create_session_for_registered_user(self, user_name):
new_id = uuid.uuid4().hex
# check if already existent (but should not be the case)
if len(DatabaseConnection.global_single_query(f"SELECT id FROM sessions WHERE id='{new_id}'")) > 0:
# okay, next try:
return self.create_session_for_registered_user(self, user_name)
query = f"INSERT INTO sessions (id, registered_user, temp_user, last_seen) VALUES ( '{new_id}', '{user_name}', NULL, '{datetime.datetime.now()}')"
DatabaseConnection.global_single_execution(query)
return new_id
def create_session_for_temp_user(self, user_name):
new_id = uuid.uuid4().hex
# check if already existent (but should not be the case)
if len(DatabaseConnection.global_single_query(f"SELECT id FROM sessions WHERE id='{new_id}'")) > 0:
# okay, next try:
return self.create_session_for_registered_user(self, user_name)
query = f"INSERT INTO sessions (id, registered_user, temp_user, last_seen) VALUES ( '{new_id}', NULL, '{user_name}', '{datetime.datetime.now()}')"
DatabaseConnection.global_single_execution(query)
return new_id
def delete_session(self, session_id):
query = f"DELETE FROM sessions WHERE id='{session_id}'"
DatabaseConnection.global_single_execution(query)
def revoke_inactive_sessions(self):
revoke_time = datetime.datetime.now() - self.session_lifespan_timedelta
query = f"SELECT * from sessions WHERE last_seen < '{get_sql_time(revoke_time)}'"
revoked_sessions = DatabaseConnection.global_single_query(query)
query = f"DELETE FROM sessions WHERE last_seen < '{get_sql_time(revoke_time)}'"
DatabaseConnection.global_single_execution(query)
return revoked_sessions