city_retrofit/hub/persistence/repositories/user.py

"""
User repository with database CRUD operations
SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Peter Yefi peteryefi@gmail.com
"""
import logging
from typing import Union, Dict
import datetime
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy import select
from hub.persistence import Repository
from hub.persistence.models import User as Model, Application as ApplicationModel, UserRoles
from hub.helpers.auth import Auth



class User(Repository):
  """
  User class
  """
  _instance = None

  def __init__(self, db_name: str, dotenv_path: str, app_env: str):
    super().__init__(db_name, dotenv_path, app_env)

  def __new__(cls, db_name, dotenv_path, app_env):
    """
    Implemented for a singleton pattern
    """
    if cls._instance is None:
      cls._instance = super(User, cls).__new__(cls)
    return cls._instance

  def insert(self, name: str, password: str, role: UserRoles, application_id: int) -> Union[Model, Dict]:
    """
    Inserts a new user
    :param name: username
    :param password: user password
    :param role: user rol [Admin or Hub_Reader]
    :param application_id: user application id
    :return: [User, Dictionary]
    """
    user = self.get_by_name_and_application(name, application_id)
    if user is None:
      try:
        user = Model(name=name, password=Auth.hash_password(password), role=role, application_id=application_id)
        self.session.add(user)
        self.session.flush()
        self.session.commit()
        return user
      except SQLAlchemyError as err:
        error_message = f'An error occurred while creating user: {err}'
        logging.error(error_message)
    return {'message': f'error creating user {name}'}

  def update(self, user_id: int, name: str, password: str, role: UserRoles) -> Union[Dict, None]:
    """
    Updates a user
    :param user_id: the id of the user to be updated
    :param name: the name of the user
    :param password: the password of the user
    :param role: the role of the user
    :return: None, Dictionary
    """
    try:
      self.session.query(Model).filter(Model.id == user_id).update({
        'name': name,
        'password': Auth.hash_password(password),
        'role': role,
        'updated': datetime.datetime.utcnow()
      })
      self.session.commit()
    except SQLAlchemyError as err:
      logging.error('Error while updating user: %s', err)
      return {'err_msg': 'Error occurred while updated user'}
    return None

  def delete(self, user_id: int):
    """
    Deletes a user with the id
    :param user_id: the user id
    :return: None
    """
    try:
      self.session.query(Model).filter(Model.id == user_id).delete()
      self.session.commit()
    except SQLAlchemyError as err:
      logging.error('Error while fetching user: %s', err)

  def get_by_name_and_application(self, name: str, application_id: int) -> Union[Model, None]:
    """
    Fetch user based on the email address
    :param name: Username
    :param application_id: User application name
    :return: User matching the search criteria or None
    """
    user = None
    try:
      user = self.session.execute(
        select(Model).where(Model.name == name, Model.application_id == application_id)
      ).first()
      if user is not None:
        user = user[0]
    except SQLAlchemyError as err:
      logging.error('Error while fetching user by name and application: %s', err)
    return user

  def get_by_name_application_id_and_password(self, name: str, password: str, application_id: int) -> Union[Model, None]:
    """
    Fetch user based on the email and password
    :param name: Username
    :param password: User password
    :param application_id: Application id
    :return: User
    """
    try:
      user = self.session.execute(
        select(Model).where(Model.name == name, Model.application_id == application_id)
      ).first()
      if user:
        if Auth.check_password(password, user[0].password):
          return user[0]
    except SQLAlchemyError as err:
      logging.error('Error while fetching user by email: %s', err)
    return None

  def get_by_name_application_uuid_and_password(self, name: str, password: str, application_uuid: str) -> Union[Model, None]:
    """
    Fetch user based on the email and password
    :param name: Username
    :param password: User password
    :param application_uuid: Application uuid
    :return: User
    """
    application = None
    try:
      application = self.session.execute(
        select(ApplicationModel).where(ApplicationModel.application_uuid == application_uuid)
      ).first()
    except SQLAlchemyError as err:
      logging.error('Error while fetching user by name: %s', err)
    return self.get_by_name_application_id_and_password(name, password, application[0].id)