80 lines
2.7 KiB
Python
80 lines
2.7 KiB
Python
from flask_restx import Resource
|
|
from app.docs.user_models import user_ns, update_user
|
|
from flask import current_app, request
|
|
from bson import json_util
|
|
from app.db.models import UserModel
|
|
from app.schemas.update_user_schema import UpdateUserRequest
|
|
from app.utils.role_required import role_required
|
|
from flask_jwt_extended import jwt_required
|
|
|
|
@user_ns.route('')
|
|
@user_ns.doc(security='Bearer Auth')
|
|
@user_ns.response(200, 'success')
|
|
class Users(Resource):
|
|
|
|
@jwt_required()
|
|
@role_required('admin')
|
|
def get(self):
|
|
user_model = UserModel()
|
|
users = user_model.list_users()
|
|
return current_app.response_class(
|
|
response=json_util.dumps({"success": True, "data": users}),
|
|
mimetype='application/json'
|
|
)
|
|
|
|
|
|
@user_ns.route('/<string:user_id>')
|
|
@user_ns.doc(security='Bearer Auth')
|
|
@user_ns.response(200, 'success')
|
|
@user_ns.response(400, 'Validation error')
|
|
class User(Resource):
|
|
def __init__(self, api=None, *args, **kwargs):
|
|
super().__init__(api, *args, **kwargs)
|
|
self.user_model = UserModel()
|
|
|
|
@jwt_required()
|
|
@role_required('admin')
|
|
def get(self, user_id):
|
|
if user := self.user_model.get_user_by_id(user_id):
|
|
|
|
return current_app.response_class(
|
|
response=json_util.dumps({"success": True, "user": user}),
|
|
mimetype='application/json'
|
|
)
|
|
|
|
return {"success": False, 'message': 'User not found'}, 404
|
|
|
|
|
|
@user_ns.expect(update_user)
|
|
@jwt_required()
|
|
@role_required('admin')
|
|
def patch(self, user_id):
|
|
data = request.get_json()
|
|
|
|
validated = UpdateUserRequest(**data)
|
|
|
|
if user := self.user_model.get_user_by_id(user_id):
|
|
user_already_exist = self.user_model.find_by_email(email=validated.email, except_user_id=user['_id'])
|
|
if user_already_exist:
|
|
return {"success": False, 'message': f'Email {validated.email} belong to another user'}, 400
|
|
else:
|
|
return {"success": False, 'message': 'User not found'}, 404
|
|
|
|
|
|
update_data = validated.model_dump(exclude_none=True)
|
|
|
|
self.user_model.update_user(user_id, update_data)
|
|
|
|
return {"success": True, 'message': f'User {user_id} updated!'}, 200
|
|
|
|
|
|
@jwt_required()
|
|
@role_required('admin')
|
|
def delete(self, user_id):
|
|
|
|
if not self.user_model.get_user_by_id(user_id):
|
|
return {"success": False, 'message': 'User not found'}, 404
|
|
|
|
self.user_model.delete_user(user_id)
|
|
|
|
return {"success": True, 'message': f'User {user_id} deleted!'}, 200 |