A multipurpose python flask API server and administration SPA
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.
 
 
 
 
 
 

115 lines
3.4 KiB

"""User API blueprint and endpoint definitions."""
from flask import Blueprint, abort, request, g
from corvus.api.decorators import return_json
from corvus.api.model import APIResponse, APIMessage, APIPage
from corvus.middleware import authentication_middleware
from corvus.model import User
from corvus.service import (
patch_service,
transformation_service,
user_service
)
from corvus.service.patch_service import get_patch_fields
from corvus.service.role_service import Role
from corvus.utility.pagination_utility import get_pagination_params
USER_BLUEPRINT = Blueprint(
name='user', import_name=__name__, url_prefix='/user')
@USER_BLUEPRINT.route('', methods=['GET'])
@return_json
@authentication_middleware.require_token_auth
@authentication_middleware.require_role(required_role=Role.USER)
def get_users() -> APIResponse:
"""
Get a list of users.
:return: a paginated list of users
"""
page, per_page = get_pagination_params(request.args)
user_page = user_service.get_users(page, per_page)
if user_page is not None:
return APIResponse(APIPage.from_page(user_page), 200)
return abort(404)
@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)