Corvus/server/corvus/api/user_api.py

"""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.middleware.authentication_middleware import Auth
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
from corvus.service.role_service import ROLE_LIST

USER_BLUEPRINT = Blueprint(
    name='user', import_name=__name__, url_prefix='/user')


@USER_BLUEPRINT.route('', methods=['GET'])
@return_json
@authentication_middleware.require(
    required_auth=Auth.TOKEN, 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(
    required_auth=Auth.TOKEN, 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(
    required_auth=Auth.TOKEN, 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(
    required_auth=Auth.TOKEN, 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(
    required_auth=Auth.TOKEN, 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)


@USER_BLUEPRINT.route('/roles', methods=['GET'])
@return_json
@authentication_middleware.require(
    required_auth=Auth.TOKEN, required_role=Role.USER)
def get_roles() -> APIResponse:
    """
    List the roles available on the service

    :return: The list of roles
    """
    return APIResponse(
        sorted({str(role.data) for role in ROLE_LIST}),
        status=200
    )