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 uuid
import datetime
import time

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


class SessionData:
  def __init__(self, session):
    self._session = session

  @property
  def city(self):
    return self._session['city']

  @property
  def id(self):
    return self._session['session_id']

  @property
  def token(self):
    return self._session['token']

  @property
  def headers(self):
    return {'session_id': str(self.id), 'token': str(self.token)}

  @property
  def greenery_percentage(self):
    return self._session['greenery_percentage']

  @greenery_percentage.setter
  def greenery_percentage(self, value):
    self._session['greenery_percentage'] = value

  @property
  def greenery_catalog(self):
    return greenery_catalog

  @property
  def construction_catalog(self):
    return construction_catalog

  @property
  def usage_catalog(self):
    return usage_catalog


def expired_sessions_collector(session_timeout_duration):
  """
  Goes through each session in sessions and removes expired session(s)
  """
  while True:
    if bool(sessions):
      for session in list(sessions):
        _expire = datetime.datetime.strptime(sessions[session]['expire'], '%Y-%m-%d %H:%M:%S.%f')
        if _expire < datetime.datetime.now():
          print("session with session_id: ", session, "expired.")
          del sessions[session]
    time.sleep(60 * int(session_timeout_duration))


def _validate_session(session_id, token, application_uuid):
  """
  Checks if session is valid
  """
  try:
    session = sessions[session_id]
    if debug_mode:
      token = session['token']
    return bool(session) and (session['token'] == token) and 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 sessions[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