77 lines
2.9 KiB
Python
77 lines
2.9 KiB
Python
"""
|
|
HeatPump Service
|
|
SPDX - License - Identifier: LGPL - 3.0 - or -later
|
|
Copyright © 2023 Project Author Peter Yefi peteryefi@gmail.com
|
|
"""
|
|
import json
|
|
from flask import Response, request
|
|
from flask_restful import Resource
|
|
from hub.hub_logger import logger
|
|
from hub_api.helpers.auth import generate_auth_token, role_required
|
|
from hub.persistence.models import UserRoles
|
|
from hub_api.config import Config
|
|
|
|
headers = {'Content-Type': 'application/json'}
|
|
|
|
|
|
class User(Resource, Config):
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
@role_required([UserRoles.Admin.value])
|
|
def post(self):
|
|
try:
|
|
payload = request.get_json()
|
|
user = self.user_factory.create_user(name=payload["name"], email=payload["email"], password=payload["password"],
|
|
role=payload["role"])
|
|
if type(user) is dict:
|
|
return Response(response=json.dumps(user), status=400, headers=headers)
|
|
return Response(response=json.dumps({'user': {'id': user.id, 'name': user.name, 'email': user.email,
|
|
'password': user.password, 'role': user.role.value}}), status=201,
|
|
headers=headers)
|
|
except Exception as err:
|
|
logger.error(err)
|
|
return Response(response=json.dumps({'err_msg': 'Sorry an error occurred while creating user'}), status=400,
|
|
headers=headers)
|
|
|
|
@role_required([UserRoles.Admin.value])
|
|
def put(self):
|
|
try:
|
|
payload = request.get_json()
|
|
res = self.user_factory.update_user(user_id=payload['id'], name=payload['name'], password=payload['password'],
|
|
role=payload['role'], email=payload['email'])
|
|
if res:
|
|
return Response(response=json.dumps(res), status=400, headers=headers)
|
|
return Response(response=json.dumps({'success': 'user updated successfully'}), status=200, headers=headers)
|
|
except Exception as err:
|
|
logger.error(err)
|
|
return Response(response=json.dumps({'err_msg': 'Sorry, an error occurred while updating user'}),
|
|
status=400, headers=headers)
|
|
|
|
|
|
class UserLogin(Resource, Config):
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
def post(self):
|
|
try:
|
|
payload = request.get_json()
|
|
user = self.ex_user_factory.login_user(email=payload["email"], password=payload["password"])
|
|
if type(user) is dict:
|
|
return Response(response=json.dumps(user), status=400, headers=headers)
|
|
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=200, headers=headers)
|
|
except Exception as err:
|
|
logger.error(err)
|
|
return Response(response=json.dumps({'err_msg': 'An error occurred while authenticating user'}), status=400)
|