nopaque/app/api/users.py

from apifairy import authenticate, body, response
from apifairy.decorators import other_responses
from flask import abort, Blueprint
from werkzeug.exceptions import InternalServerError
from app.email import create_message, send
from app import db
from app.models import User
from .auth import auth_error_responses, token_auth
from .schemas import EmptySchema, UserSchema


bp = Blueprint('users', __name__)
user_schema = UserSchema()
users_schema = UserSchema(many=True)


@bp.route('', methods=['GET'])
@authenticate(token_auth, role='Administrator')
@response(users_schema)
@other_responses(auth_error_responses)
def get_users():
    """Get all users"""
    return User.query.all()


@bp.route('', methods=['POST'])
@body(user_schema)
@response(user_schema, 201)
@other_responses({InternalServerError.code: InternalServerError.description})
def create_user(args):
    """Create a new user"""
    try:
        user = User.create(
            email=args['email'].lower(),
            password=args['password'],
            username=args['username']
        )
    except OSError:
        abort(500)
    msg = create_message(
        user.email,
        'Confirm Your Account',
        'auth/email/confirm',
        token=user.generate_confirm_token(),
        user=user
    )
    send(msg)
    db.session.commit()
    return user, 201


@bp.route('/<hashid:user_id>', methods=['DELETE'])
@authenticate(token_auth)
@response(EmptySchema, status_code=204)
@other_responses(auth_error_responses)
def delete_user(user_id):
    """Delete a user by id"""
    current_user = token_auth.current_user()
    user = User.query.get(user_id)
    if user is None:
        abort(404)
    if not (user == current_user or current_user.is_administrator):
        abort(403)
    user.delete()
    db.session.commit()
    return {}, 204


@bp.route('/<hashid:user_id>', methods=['GET'])
@authenticate(token_auth)
@response(user_schema)
@other_responses(auth_error_responses)
@other_responses({404: 'User not found'})
def get_user(user_id):
    """Retrieve a user by id"""
    current_user = token_auth.current_user()
    user = User.query.get(user_id)
    if user is None:
        abort(404)
    if not (user == current_user or current_user.is_administrator):
        abort(403)
    return user


@bp.route('/<username>', methods=['GET'])
@authenticate(token_auth)
@response(user_schema)
@other_responses(auth_error_responses)
@other_responses({404: 'User not found'})
def get_user_by_username(username):
    """Retrieve a user by username"""
    current_user = token_auth.current_user()
    user = User.query.filter(User.username == username).first()
    if user is None:
        abort(404)
    if not (user == current_user or current_user.is_administrator):
        abort(403)
    return user