api_v1.4/hub_api/user.py

"""
HeatPump Service
SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Project Author Peter Yefi peteryefi@gmail.com
"""
import json
from flask import Response
from flask_apispec import use_kwargs, doc
from flask_apispec.views import MethodResource
from flask_restful import Resource
from marshmallow import Schema, fields
from imports.user_factory import UserFactory
from exports.user_factory import UserFactory as ExUserFactory
import os
from hub_logger import logger
from hub_api.helpers.auth import generate_auth_token, role_required
from persistence.models import UserRoles


class AuthorizationHeader(Schema):
  Authorization = fields.Str(required=True, description='Authorization token')
  AppID = fields.Str(required=True, description='ID of app accessing API')


class LoginPostSchema(Schema):
  """
  Defines post data for users
  """
  password = fields.String(required=True, description='Password of user')
  email = fields.String(required=True, description='Email of user')


class UserPostSchema(LoginPostSchema):
  """
  Defines post data for users
  """
  name = fields.String(required=True, description='Name of user')
  role = fields.String(required=True, description='Allowed user roles', enum=['Admin', 'Hub_Reader'])


class UserPutSchema(UserPostSchema):
  """
  Defines put data for users
  """
  id = fields.Int(required=True, description='The Id of the user to be Updated')


class User(MethodResource, Resource):
  def __init__(self):
    self.user_factory = UserFactory(db_name='hub_prod', app_env='PROD',
                                    dotenv_path="{}/.env".format(os.path.expanduser('~')))

  @doc(description='Create users', tags=['CreateUser'])
  @role_required(UserRoles.Admin.value)
  @use_kwargs(AuthorizationHeader, location='headers')
  @use_kwargs(UserPostSchema)
  def post(self, **kwargs):
    try:

      user = self.user_factory.create_user(name=kwargs["name"], email=kwargs["email"], password=kwargs["password"],
                                           role=kwargs["role"])
      if type(user) is dict:
        return Response(response=json.dumps(user), status=400)
      return Response(response=json.dumps({'user': {'id': user.id, 'name': user.name, 'email': user.email,
                                                    'password': user.password, 'role': user.role.value}}), status=201)
    except Exception as err:
      logger.error(err)
      return Response(response=json.dumps({'err_msg': 'Sorry an error occurred while creating user'}), status=400)

  @doc(description='Get all users', tags=['UpdateUsers'])
  @use_kwargs(UserPutSchema)
  @role_required(UserRoles.Admin.value)
  @use_kwargs(AuthorizationHeader, location='headers')
  def put(self, **kwargs):
    try:
      res = self.user_factory.update_user(user_id=kwargs['id'], name=kwargs['name'], password=kwargs['password'],
                                          role=kwargs['role'], email=kwargs['email'])
      if res:
        return Response(response=json.dumps(res), status=400)
      return Response(response=json.dumps({'success': 'user updated successfully'}), status=200)
    except Exception as err:
      logger.error(err)
      return Response(response=json.dumps({'err_msg': 'Sorry, an error occurred while updating user'}),
                      status=400)


class UserLogin(MethodResource, Resource):
  def __init__(self):
    self.user_factory = ExUserFactory(db_name='hub_prod', app_env='PROD',
                                      dotenv_path="{}/.env".format(os.path.expanduser('~')))

  @doc(description='Create users', tags=['LoginUser'])
  @use_kwargs(LoginPostSchema)
  def post(self, **kwargs):
    try:
      user = self.user_factory.login_user(email=kwargs["email"], password=kwargs["password"])
      if type(user) is dict:
        return Response(response=json.dumps(user), status=400)
      user = user[0]
      user_dict = {
        'user': {
          'id': user.id,
          'name': user.name,
          'email': user.email,
          'password': user.password,
          'role': user.role.value,
        }
      }
      user_dict['token'] = generate_auth_token(user_dict)
      return Response(response=json.dumps(user_dict), status=201)
    except Exception as err:
      logger.error(err)
      return Response(response=json.dumps({'err_msg': 'Sorry an error occurred while authenticating user'}), status=400)