api_v1.4/hub_api/helpers/session_helper.py

"""
Session helper
SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Project Author name guillermo.gutierrezmorote@concordia.ca
"""
import datetime
import shutil
import time
import uuid
from copy import deepcopy
from pathlib import Path

sessions = {}
begin_time = None
swagger_data = None
city = None
greenery_catalog = None
construction_catalog = None
usage_catalog = None
debug_mode = False


def expired_sessions_collector(session_timeout_duration):
  """
  Goes through each session in sessions and removes expired session(s)
  """
  while True:
    if bool(sessions):
      _sessions = deepcopy(sessions)
      for session_uuid in _sessions:
        _expire = datetime.datetime.strptime(_sessions[session_uuid]['expire'], '%Y-%m-%d %H:%M:%S.%f')
        if _expire < datetime.datetime.now():
          print("session for user: ", _sessions[session_uuid]['user'], "expired.")
          response_path = (Path(__file__).parent.parent / f'response_files/{session_uuid}').resolve()
          shutil.rmtree(response_path)
          del sessions[session_uuid]
    time.sleep(60 * int(session_timeout_duration))


def _validate_session(session_id, token, application_uuid):
  """
  Checks if session is valid
  """
  try:
    _session = session(session_id)
    if debug_mode:
      token = _session['token']
    return bool(_session) and (_session['token'] == token) and str(_session['application_uuid']) == application_uuid
  except KeyError:
    return False


def remove_session(session_id, token, application_uuid):
  """
  Remove a session from the sessions array
  """
  if _validate_session(session_id, token, application_uuid):
    del sessions[session_id]
    return True
  return False


def refresh_session(session_id, token, application_uuid):
  """
  Validate and extend current session
  :return: valid session
  """
  if _validate_session(session_id, token, application_uuid):
    sessions[session_id]['expire'] = str(datetime.datetime.now() + datetime.timedelta(minutes=5))
    sessions[session_id]['token'] = str(uuid.uuid4())
    return session(session_id)
  return None


def active_session(session_id, token, application_uuid):
  _is_valid = _validate_session(session_id=session_id, token=token, application_uuid=application_uuid)
  if _is_valid:
    refresh_session(session_id, token, application_uuid)
  return _is_valid


def session(session_id) -> {}:
  return sessions[session_id]