users get, post, delete, update impl

Author: msnodeve

Pipfile

@@ -7,6 +7,7 @@ verify_ssl = true
 pytest = "*"
 pytest-cov = "*"
 pylint = "*"
+autopep8 = "*"
 
 [packages]
 flask = "*"
@@ -16,6 +17,7 @@ flask-migrate = "*"
 pymysql = "*"
 flask-restplus = "*"
 flask-marshmallow = "*"
+marshmallow-jsonapi = "*"
 
 [requires]
 python_version = "3.7"

Pipfile.lock

@@ -1,41 +1,19 @@
 """
     app init
 """
-
 from flask import Flask, render_template, jsonify
 from flask_restplus import Resource, Api, fields, reqparse
 from flask_sqlalchemy import SQLAlchemy
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.sql import text
 from flask_marshmallow import Marshmallow
+from app.api.database import DB
+from app.users.views import API
 
-DB = SQLAlchemy()
 SQLALCHEMY_DATABASE_URI = \
     ("mysql+pymysql://{USER}:{PASSWORD}@{ADDR}:{PORT}/{NAME}?charset=utf8")
-# API Doc에 대한 설명을 덧 붙이기 위한 생성자
-API = Api(version='1.0', title='Board API',
-    description='A simple Board RESTPlus API',)
 # 설명할 API에 대한 것
-api_ns = API.namespace('userinfo', description='UserInfo operations')
 MA = Marshmallow()
-
-class UserInfo(DB.Model):
-    """ UserInfo model """
-    __tablename__ = "userinfo"
-    __table_args__ = {'mysql_collate' : 'utf8_general_ci'}
-    id = DB.Column("id", DB.Integer, primary_key=True)
-    name = DB.Column("name", DB.String(250), nullable=False)
-    age = DB.Column("age", DB.Integer, nullable=False)
-    tel = DB.Column("tel", DB.String(20), nullable=False)
-    email = DB.Column("email", DB.String(50), nullable=False)
-    created = DB.Column(DB.TIMESTAMP, server_default=text("CURRENT_TIMESTAMP"), nullable=False)
-
-    def __init__(self, name, age, tel, email):
-        self.name = name
-        self.age = age
-        self.tel = tel
-        self.email = email
-
 def create_app() -> (Flask):
     """ create_app() 함수를 호출해 앱을 초기화 """
 
@@ -58,116 +36,5 @@ def create_app() -> (Flask):
     API.init_app(app)
     MA.init_app(app)
 
-    """ API model을 보이기 위한 설명 추가 """
-    # 나중에 따로 빼야 할 것 같다.
-    user_info = API.model('UserInfo', {
-        'name' : fields.String(required=True, description="User's name"),
-        'age' : fields.Integer(required=True, description="User's age"),
-        'tel' : fields.String(required=True, description="User's tel"),
-        'email' : fields.String(required=True, description="User's email")
-    })
-
-    class UserSchema(MA.Schema):
-        class Meta:
-            # Fields to expose
-            fields = ("name", "age", "tel", "email")
-
-
-
-    """ API.route part"""
-    @api_ns.route("/hello")
-    class HelloWorld(Resource):
-        def get(self):
-            return {'hello': 'world'}
-
-    @api_ns.route('/users')
-    class Show(Resource):
-        post_parser = reqparse.RequestParser(bundle_errors=True)
-        post_parser.add_argument('name', required=True, type=str, help="user name",location='json')
-        post_parser.add_argument('age', required=True, type=int, help="user age",location='json')
-        post_parser.add_argument('tel', required=True, type=str, help="user tel",location='json')
-        post_parser.add_argument('email', required=True, type=str, help="user email",location='json')
-        def get(self):
-            user_schema = UserSchema(many=True)
-            all_users = DB.session.query(UserInfo).all()
-            result = user_schema.dump(all_users)
-            return jsonify(result.data)
-
-        @api_ns.expect(user_info)
-        def post(self):
-            args_ = self.post_parser.parse_args()
-            user = UserInfo(name=args_['name'], age=args_['age'], tel=args_['tel'], email=args_['email'])
-            message = "not working"
-            try:
-                DB.session.add(user)
-                DB.session.commit()
-                message = "success"
-            except SQLAlchemyError as err:
-                message = str(err)
-                DB.session.rollback()
-            finally:
-                DB.session.close()
-            return message
-
-        @api_ns.expect(user_info)
-        def delete(self):
-            args_ = self.post_parser.parse_args()
-            user = UserInfo.query.filter_by(name=args_['name']).first()
-            try:
-                DB.session.delete(user)
-                DB.session.commit()
-                message = "success"
-            except SQLAlchemyError as err:
-                message = str(err)
-                DB.session.rollback()
-            finally:
-                DB.session.close()
-            return message
-
-    @api_ns.route('/test')
-    class Test(Resource):
-        post_parser = reqparse.RequestParser(bundle_errors=True)
-        post_parser.add_argument('name', required=True, type=str, help="user name",location='json')
-        post_parser.add_argument('age', required=True, type=int, help="user age",location='json')
-        post_parser.add_argument('tel', required=True, type=str, help="user tel",location='json')
-        post_parser.add_argument('email', required=True, type=str, help="user email",location='json')
-        def get(self):
-            return {'test' : 'test'}
-        
-        @api_ns.expect(user_info)
-        def post(self):
-            args_ = self.post_parser.parse_args()
-            user = UserInfo(name=args_['name'], age=args_['age'], tel=args_['tel'], email=args_['email'])
-            return {'name' : 'server' + user.name, 'age' : user.age, 'tel' : user.tel, 'email' : user.email}
-
-
-    @api_ns.route('/test')
-    class HelloWorld(Resource):
-        post_parser = reqparse.RequestParser(bundle_errors=True)
-        post_parser.add_argument('name', required=True, type=str, help="user name",location='json')
-        post_parser.add_argument('tel', type=str, help="user tel",location='json')
-
-        def get(self):
-            userInfo = DB.session.query(UserInfo).all()
-            import pdb; pdb.set_trace()
-            return {'hello': 'world'}
-
-        @api_ns.expect(user_info)
-        def post(self):
-            """ 유저 생성 """
-            args_ = self.post_parser.parse_args()
-            user = UserInfo(name=args_['name'], tel=args_['tel'])
-            message = "not working"
-            try:
-                DB.session.add(user)
-                DB.session.commit()
-                message = "success"
-            except SQLAlchemyError as err:
-                message = str(err)
-                DB.session.rollback()
-            finally:
-                DB.session.close()
-            return message
-
     """ return part """
-    return app
+    return app

app/__init__.py

@@ -1,6 +1,62 @@
 from flask import jsonify
+from flask import make_response
 from flask_sqlalchemy import SQLAlchemy
 from sqlalchemy.exc import IntegrityError
 from marshmallow import ValidationError
 from app.constants import STATUS_CODE
-from flask_restplus import reqparse
+from flask_restplus import reqparse
+
+DB = SQLAlchemy()
+
+class CRUD:
+    body = ''
+    status_code = STATUS_CODE.NOT_IMPLEMENTED
+    def add(self, resource, schema):
+        try:
+            DB.session.add(resource)
+            DB.session.commit()
+            query = resource.query.get(resource.id)
+            self.body = jsonify(schema.dump(resource).data)
+            self.status_code = STATUS_CODE.CREATED
+        except IntegrityError as error:
+            DB.session.rollback()
+            error_message = str(error)
+            self.body = jsonify({"error": error_message, "type":"IntegrityError"})
+            if "Duplicate entry" in error_message:
+                self.status_code = 404
+            else:
+                self.status_code = 400
+        finally:
+            response = (self.body, self.status_code.value)
+            response = make_response(response)
+            
+        return response 
+
+    def update(self, args, schema):
+        try:
+            for key, value in args.items():
+                setattr(self, key, value)
+            DB.session.commit()
+            self.body = jsonify(schema.dump(self).data)
+            self.status_code = STATUS_CODE.OK
+        except IntegrityError as error:
+            DB.session.rollback()
+            error_message = str(error) 
+            self.body = jsonify({"error": error_message, "type":"IntegrityError"})
+            if "Duplicate entry" in error_message:
+                self.status_code = STATUS_CODE.CONFLICT
+            else:
+                self.status_code = STATUS_CODE.BAD_REQUEST
+        finally:
+            response = (self.body, self.status_code.value)
+            response = make_response(response)
+        return response
+
+    def delete(self, resource, schema):
+        DB.session.delete(resource)
+        DB.session.commit()
+        self.body = jsonify({"message":"success"})
+        self.status_code = STATUS_CODE.OK
+        response = (self.body, self.status_code.value)
+        response = make_response(response)
+        return response

app/api/database.py

@@ -0,0 +1,41 @@
+from enum import Enum
+
+class STATUS_CODE(Enum):
+	OK = 200
+	CREATED = 201
+	NO_CONTENT = 204
+	BAD_REQUEST = 400
+	UNAUTHORIZED = 401
+	FORBIDDEN = 403
+	NOT_FOUND = 404
+	METHOD_NOT_ALLOWED = 405
+	CONFLICT = 409	
+	INTERNAL_SERVER_ERROR = 500
+	NOT_IMPLEMENTED = 501
+	BAD_GATEWAY = 502
+
+responses = {item.value: item.name for item in STATUS_CODE}
+
+def support_codes(unsupport_codes):
+	unsupport_codes = [code.value for code in unsupport_codes]
+	return {code: name for code, name in responses.items() if code not in unsupport_codes}
+	
+GET = support_codes(unsupport_codes=[
+						STATUS_CODE.CREATED, STATUS_CODE.NO_CONTENT,
+						STATUS_CODE.CONFLICT,
+						STATUS_CODE.METHOD_NOT_ALLOWED,
+						STATUS_CODE.NOT_IMPLEMENTED])
+POST = support_codes(unsupport_codes=[
+						STATUS_CODE.OK, STATUS_CODE.NO_CONTENT,
+						STATUS_CODE.METHOD_NOT_ALLOWED,
+						STATUS_CODE.NOT_IMPLEMENTED])
+PATCH = support_codes(unsupport_codes=[
+						STATUS_CODE.CREATED,
+						STATUS_CODE.NO_CONTENT,
+						STATUS_CODE.METHOD_NOT_ALLOWED,
+						STATUS_CODE.NOT_IMPLEMENTED])
+DELETE = support_codes(unsupport_codes=[
+						STATUS_CODE.CREATED,
+						STATUS_CODE.CONFLICT,
+						STATUS_CODE.METHOD_NOT_ALLOWED,
+						STATUS_CODE.NOT_IMPLEMENTED])

app/constants.py

@@ -0,0 +1,42 @@
+from app.api.database import DB, CRUD
+from marshmallow_jsonapi import Schema, fields
+from flask_sqlalchemy import SQLAlchemy
+from marshmallow import validate
+from sqlalchemy.sql import text
+
+
+class Users(DB.Model, CRUD):
+    __tablename__ = 'users'
+    __table_args__ = {'mysql_collate': 'utf8_general_ci'}
+
+    id = DB.Column(DB.Integer, primary_key=True)
+    name = DB.Column(DB.String(255), nullable=False)
+    email = DB.Column(DB.String(50), unique=True, nullable=False)
+    password = DB.Column(DB.String(255), nullable=False)
+    created = DB.Column(DB.TIMESTAMP, server_default=text(
+        "CURRENT_TIMESTAMP"), nullable=False)
+
+    def __init__(self, name, email, password):
+        self.name = name
+        self.email = email
+        self.password = password
+
+
+class UsersSchema(Schema):
+    not_blank = validate.Length(min=1, error='Field cannot be blank')
+    id = fields.Integer(dump_only=True)
+    email = fields.String(validate=not_blank)
+    name = fields.String(validate=not_blank)
+    password = fields.String(validate=not_blank)
+    created = fields.String(validate=not_blank)
+
+    def get_top_level_links(self, data, many):
+        if many:
+            self_link = "/users"
+        else:
+            self_link = "/users/{}".format(data['id'])
+        return {'self': self_link}
+
+    class Meta:
+        type_ = 'users'
+        strict = True