You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
96 lines
2.9 KiB
96 lines
2.9 KiB
"""User API blueprint and endpoint definitions."""
|
|
from flask import Blueprint, abort, request, g
|
|
|
|
from atheneum.api.decorators import return_json
|
|
from atheneum.api.model import APIResponse, APIMessage
|
|
from atheneum.middleware import authentication_middleware
|
|
from atheneum.model import User
|
|
from atheneum.service import (
|
|
patch_service,
|
|
transformation_service,
|
|
user_service
|
|
)
|
|
from atheneum.service.patch_service import get_patch_fields
|
|
from atheneum.service.role_service import Role
|
|
|
|
USER_BLUEPRINT = Blueprint(
|
|
name='user', import_name=__name__, url_prefix='/user')
|
|
|
|
|
|
@USER_BLUEPRINT.route('/<name>', methods=['GET'])
|
|
@return_json
|
|
@authentication_middleware.require_token_auth
|
|
@authentication_middleware.require_role(required_role=Role.USER)
|
|
def get_user(name: str) -> APIResponse:
|
|
"""
|
|
Get a user.
|
|
|
|
:return: user if exists, else 404
|
|
"""
|
|
user = user_service.find_by_name(name)
|
|
if user is not None:
|
|
return APIResponse(user, 200)
|
|
return abort(404)
|
|
|
|
|
|
@USER_BLUEPRINT.route('/<name>', methods=['PATCH'])
|
|
@return_json
|
|
@authentication_middleware.require_token_auth
|
|
@authentication_middleware.require_role(required_role=Role.USER)
|
|
def patch_user(name: str) -> APIResponse:
|
|
"""
|
|
Patch a user.
|
|
|
|
:return: user if patched, 4xx error on patching issue, 404 on nonexistent
|
|
"""
|
|
user = user_service.find_by_name(name)
|
|
if user is not None:
|
|
user_patch: User = transformation_service.deserialize_model(
|
|
User, request.json)
|
|
patched_user = patch_service.patch(
|
|
g.user, user, user_patch, get_patch_fields(request.json))
|
|
return APIResponse(patched_user, 200)
|
|
return abort(404)
|
|
|
|
|
|
@USER_BLUEPRINT.route('/', methods=['POST'])
|
|
@return_json
|
|
@authentication_middleware.require_token_auth
|
|
@authentication_middleware.require_role(required_role=Role.ADMIN)
|
|
def register_user() -> APIResponse:
|
|
"""
|
|
Register a user with the service.
|
|
|
|
:return: The newly registered User
|
|
"""
|
|
new_user: User = transformation_service.deserialize_model(
|
|
User, request.json)
|
|
|
|
requested_password = None
|
|
if 'password' in request.json:
|
|
requested_password = request.json['password'].strip()
|
|
|
|
registered_user = user_service.register(
|
|
name=new_user.name,
|
|
password=requested_password,
|
|
role=new_user.role
|
|
)
|
|
return APIResponse(payload=registered_user, status=200)
|
|
|
|
|
|
@USER_BLUEPRINT.route('/<name>', methods=['DELETE'])
|
|
@return_json
|
|
@authentication_middleware.require_token_auth
|
|
@authentication_middleware.require_role(required_role=Role.ADMIN)
|
|
def delete_user(name: str) -> APIResponse:
|
|
"""
|
|
Delete a user with the service.
|
|
|
|
:return: The newly registered User
|
|
"""
|
|
user = user_service.find_by_name(name)
|
|
if user is not None:
|
|
user_service.delete(user)
|
|
return APIResponse(
|
|
APIMessage(True, 'Successfully Deleted'), status=200)
|
|
return abort(404)
|