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

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):
  # loop through all sessions and remove expired sessions
  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):
  try:
    if bool(sessions[session_id]) and sessions[session_id]['token'] == token and \
            sessions[session_id]['application_uuid'] == application_uuid:
      return True
  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):
  return _validate_session(session_id = session_id, token = token, application_uuid = application_uuid)