backend/src/resources.py

#!/usr/bin/env python3
import datetime

from flask_jwt_extended import create_access_token, jwt_required
from flask_restful import Resource
from flask import request, current_app, abort

from db import db
from models import User
from schemas import UserSchema, UserMetadataSchema
from config import REGISTER_DISABLED

"""
Flask Restful endpoints
"""

__author__ = '@tormakris'
__copyright__ = "Copyright 2020, videON Team"
__module_name__ = "resources"
__version__text__ = "1"


class SignupApi(Resource):
    """
    See: https://swagger.kmlabz.com/?urls.primaryName=videON%20Backend#/backend/createuser
    """

    userschema = UserSchema(many=False)
    usermetadataschema = UserMetadataSchema(many=False)

    def post(self):
        if REGISTER_DISABLED:
            abort(401, "register disabled")

        body = request.get_json()

        try:
            userobj = self.userschema.load(body)
        except Exception as e:
            current_app.logger.warning(e)
            abort(417, "invalid json schema")

        user = User(name=userobj['name'], password=userobj['password'])
        try:
            user.hash_password()
            db.session.add(user)
            db.session.commit()
        except Exception as e:
            db.session.rollback()
            current_app.logger.warning(e)
            abort(503, "user already exists")

        return self.usermetadataschema.dump(user), 200


class LoginApi(Resource):
    """
    See: https://swagger.kmlabz.com/?urls.primaryName=videON%20Backend#/backend/logon
    """

    userschema = UserSchema(many=False)
    usermetadataschema = UserMetadataSchema(many=False)

    def post(self):
        body = request.get_json()

        try:
            userobj = self.userschema.load(body)
        except Exception as e:
            current_app.logger.warning(e)
            abort(417, "invalid json schema")

        user = User.query.filter_by(name=userobj['name']).first_or_404()
        authorized = user.check_password(userobj['password'])
        if not authorized:
            abort(401, "username or password incorrect")

        try:
            user.last_logon = datetime.datetime.now()
            db.session.add(user)
            db.session.commit()
        except Exception as e:
            db.session.rollback()
            current_app.logger.exception(e)
            abort(503, "db session error")

        expires = datetime.timedelta(days=7)
        access_token = create_access_token(identity=str(user.name), expires_delta=expires)
        return {'token': access_token}, 200


class UsersApi(Resource):
    """
    See: https://swagger.kmlabz.com/?urls.primaryName=videON%20Backend#/backend/getall
    """
    usermetadataschema = UserMetadataSchema(many=True)

    def get(self):
        users = User.query.all()
        return self.usermetadataschema.dump(users), 200


class UserParameterApi(Resource):
    userschema = UserSchema(many=False)
    usermetadataschema = UserMetadataSchema(many=False)

    def get(self, username: str):
        """
        See: https://swagger.kmlabz.com/?urls.primaryName=videON%20Backend#/backend/getauser
        :param username: Username of user (url parameter)
        :return:
        """
        user = User.query.filter_by(name=username).first_or_404()
        return self.usermetadataschema.dump(user), 200

    def delete(self, username: str):
        """
        See: https://swagger.kmlabz.com/?urls.primaryName=videON%20Backend#/backend/deleteuser
        :param username: Username of user (url parameter)
        :return:
        """
        user = User.query.filter_by(name=username).first_or_404()

        try:
            db.session.delete(user)
            db.session.commit()
        except Exception as e:
            db.session.rollback()
            current_app.logger.exception(e)
            abort(503, "db session error")

        return self.usermetadataschema.dump(user), 200

    def put(self, username: str):
        """
        See: https://swagger.kmlabz.com/?urls.primaryName=videON%20Backend#/backend/modifyUser
        :param username: Username of user (url parameter)
        :return:
        """
        body = request.get_json()
        user = User.query.filter_by(name=username).first_or_404()
        try:
            userobj = self.userschema.load(body)
        except Exception as e:
            current_app.logger.warning(e)
            abort(417, "invalid json schema")

        try:
            user.password = userobj['password']
            user.hash_password()
            db.session.commit()
        except Exception as e:
            db.session.rollback()
            current_app.logger.exception(e)
            abort(503, "db session error")

        return self.usermetadataschema.dump(user), 200