加入webargs验证扩展,分区用于验证的inschema和用户序列化的outschema

applications/common/admin.py

@@ -5,7 +5,7 @@ from flask import session, make_response, current_app
 from flask_login import current_user
 
 from applications.common.utils.gen_captcha import gen_captcha
-from applications.schemas import PowerSchema
+from applications.schemas import PowerOutSchema
 
 
 # 授权路由存入session
@@ -39,7 +39,7 @@ def make_menu_tree():
             if int(p.type) == 0 or int(p.type) == 1:
                 powers.append(p)
 
-    power_schema = PowerSchema(many=True)  # 用已继承 ma.ModelSchema 类的自定制类生成序列化类
+    power_schema = PowerOutSchema(many=True)  # 用已继承 ma.ModelSchema 类的自定制类生成序列化类
     power_dict = power_schema.dump(powers)  # 生成可序列化对象
     power_dict.sort(key=lambda x: x['id'], reverse=True)
 

applications/common/utils/upload.py

@@ -4,14 +4,14 @@ from sqlalchemy import desc
 from applications.extensions import db
 from applications.extensions.init_upload import photos
 from applications.models import Photo
-from applications.schemas import PhotoSchema
+from applications.schemas import PhotoOutSchema
 from applications.common.curd import model_to_dicts
 
 
 def get_photo(page, limit):
     photo = Photo.query.order_by(desc(Photo.create_time)).paginate(page=page, per_page=limit, error_out=False)
     count = Photo.query.count()
-    data = model_to_dicts(schema=PhotoSchema, data=photo.items)
+    data = model_to_dicts(schema=PhotoOutSchema, data=photo.items)
     return data, count
 
 

applications/extensions/init_error_views.py

@@ -1,4 +1,4 @@
-from flask import render_template
+from flask import render_template, jsonify
 
 
 def init_error_views(app):
@@ -13,3 +13,22 @@ def init_error_views(app):
     @app.errorhandler(500)
     def internal_server_error(e):
         return render_template('errors/500.html'), 500
+
+    # Return validation errors as JSON
+    @app.errorhandler(422)
+    @app.errorhandler(400)
+    def handle_error(err):
+        headers = err.data.get("headers", None)
+        messages = err.data.get("messages", ["Invalid request."]).get('json')
+        print(err.data.get("messages"))
+        print(messages.items())
+        msg = ''
+
+        for i in messages.items():
+            msg = str(i[0]) + str(i[1][0])
+            break
+
+        if headers:
+            return jsonify({"success": False, "msg": msg})
+        else:
+            return jsonify({"success": False, "msg": msg})

applications/extensions/init_webargs.py

@@ -0,0 +1,9 @@
+from marshmallow import EXCLUDE
+from webargs.flaskparser import FlaskParser
+
+
+class Parser(FlaskParser):
+    DEFAULT_UNKNOWN_BY_LOCATION = {"query": EXCLUDE}
+
+
+parser = Parser()

applications/schemas/__init__.py

@@ -1,7 +1,7 @@
-from .admin_user import UserSchema
-from .admin_role import RoleSchema
-from .admin_power import PowerSchema, PowerSchema2
-from .admin_dict import DictDataSchema, DictTypeSchema
-from .admin_dept import DeptSchema
-from .admin_log import LogSchema
-from .admin_photo import PhotoSchema
+from .admin_user import UserOutSchema
+from .admin_role import RoleOutSchema
+from .admin_power import PowerOutSchema, PowerOutSchema2
+from .admin_dict import DictDataOutSchema, DictTypeOutSchema
+from .admin_dept import DeptOutSchema
+from .admin_log import LogOutSchema
+from .admin_photo import PhotoOutSchema

applications/schemas/admin_dept.py

@@ -2,7 +2,18 @@ from applications.extensions import  ma
 from marshmallow import fields, validate
 
 
-class DeptSchema(ma.Schema):
+class DeptInSchema(ma.Schema):
+    parentId = fields.Integer(required=True)
+    deptName = fields.Str(required=True)
+    leader = fields.Str(required=True)
+    phone = fields.Str(required=True)
+    email = fields.Str(validate=validate.Email())
+    address = fields.Str()
+    status = fields.Str(validate=validate.OneOf(["0", "1"]))
+    sort = fields.Integer()
+
+
+class DeptOutSchema(ma.Schema):
     deptId = fields.Integer(attribute="id")
     parentId = fields.Integer(attribute="parent_id")
     deptName = fields.Str(attribute="dept_name")

applications/schemas/admin_dict.py

@@ -2,7 +2,7 @@ from applications.extensions import ma
 from marshmallow import fields
 
 
-class DictTypeSchema(ma.Schema):
+class DictTypeOutSchema(ma.Schema):
     id = fields.Str(attribute="id")
     typeName = fields.Str(attribute="type_name")
     typeCode = fields.Str(attribute="type_code")
@@ -13,7 +13,7 @@ class DictTypeSchema(ma.Schema):
     enable = fields.Str()
 
 
-class DictDataSchema(ma.Schema):
+class DictDataOutSchema(ma.Schema):
     dataId = fields.Str(attribute="id")
     dataLabel = fields.Str(attribute="data_label")
     dataValue = fields.Str(attribute="data_value")

applications/schemas/admin_log.py

@@ -2,7 +2,7 @@ from applications.extensions import ma
 from marshmallow import fields
 
 
-class LogSchema(ma.Schema):
+class LogOutSchema(ma.Schema):
     id = fields.Integer()
     method = fields.Str()
     uid = fields.Str()

applications/schemas/admin_photo.py

@@ -2,7 +2,7 @@ from applications.extensions import ma
 from marshmallow import fields
 
 
-class PhotoSchema(ma.Schema):
+class PhotoOutSchema(ma.Schema):
     id = fields.Integer()
     name = fields.Str()
     href = fields.Str()

applications/schemas/admin_power.py

@@ -3,7 +3,7 @@ from marshmallow import fields
 
 
 # 权限models序列化类
-class PowerSchema(ma.Schema):
+class PowerOutSchema(ma.Schema):
     id = fields.Integer()
     title = fields.Str(attribute="name")
     type = fields.Str()
@@ -18,7 +18,7 @@ class PowerSchema(ma.Schema):
     enable = fields.Integer()
 
 
-class PowerSchema2(ma.Schema):  # 序列化类
+class PowerOutSchema2(ma.Schema):  # 序列化类
     powerId = fields.Str(attribute="id")
     powerName = fields.Str(attribute="name")
     powerType = fields.Str(attribute="type")

applications/schemas/admin_role.py

@@ -2,7 +2,7 @@ from applications.extensions import ma
 from marshmallow import fields
 
 
-class RoleSchema(ma.Schema):
+class RoleOutSchema(ma.Schema):
     id = fields.Integer()
     roleName = fields.Str(attribute="name")
     roleCode = fields.Str(attribute="code")

applications/schemas/admin_user.py

@@ -4,7 +4,7 @@ from applications.models import Dept
 
 
 # 用户models的序列化类
-class UserSchema(ma.Schema):
+class UserOutSchema(ma.Schema):
     id = fields.Integer()
     username = fields.Str()
     realname = fields.Str()

applications/view/admin/admin_log.py

@@ -3,7 +3,7 @@ from sqlalchemy import desc
 from applications.common.utils.http import table_api
 from applications.common.utils.rights import authorize
 from applications.models import AdminLog
-from applications.schemas import LogSchema
+from applications.schemas import LogOutSchema
 from applications.common.curd import model_to_dicts
 
 admin_log = Blueprint('adminLog', __name__, url_prefix='/admin/log')
@@ -24,7 +24,7 @@ def login_log():
     # 使用分页获取data需要.items
     log = AdminLog.query.filter_by(url='/passport/login').order_by(desc(AdminLog.create_time)).layui_paginate()
     count = log.total
-    return table_api(data= model_to_dicts(schema=LogSchema, data=log.items), count=count)
+    return table_api(data= model_to_dicts(schema=LogOutSchema, data=log.items), count=count)
 
 
 # 操作日志
@@ -37,4 +37,4 @@ def operate_log():
         AdminLog.url != '/passport/login').order_by(
         desc(AdminLog.create_time)).layui_paginate()
     count = log.total
-    return table_api(data=model_to_dicts(schema=LogSchema, data=log.items), count=count)
+    return table_api(data=model_to_dicts(schema=LogOutSchema, data=log.items), count=count)

applications/view/admin/curd.py

@@ -1,8 +0,0 @@
-from flask import Blueprint
-
-admin_curd = Blueprint('adminCurd', __name__, url_prefix='/admin/curd')
-
-
-@admin_curd.route('/')
-def index():
-    return "功能开发中"

applications/view/admin/dict.py

@@ -7,7 +7,7 @@ from applications.common.utils.rights import authorize
 from applications.common.utils.validate import xss_escape
 from applications.extensions import db
 from applications.models import DictType, DictData
-from applications.schemas import DictTypeSchema, DictDataSchema
+from applications.schemas import DictTypeOutSchema, DictDataOutSchema
 
 admin_dict = Blueprint('adminDict', __name__, url_prefix='/admin/dict')
 
@@ -32,7 +32,7 @@ def dict_type_data():
     # 使用分页获取data需要.items
     dict_all = DictType.query.filter(mf.get_filter(DictType)).layui_paginate()
     count = dict_all.total
-    data = curd.model_to_dicts(schema=DictTypeSchema, data=dict_all.items)
+    data = curd.model_to_dicts(schema=DictTypeOutSchema, data=dict_all.items)
     return table_api(data=data, count=count)
 
 
@@ -129,7 +129,7 @@ def dict_code_data():
     type_code = xss_escape(request.args.get('typeCode', type=str))
     dict_data = DictData.query.filter_by(type_code=type_code).layui_paginate()
     count = dict_data.total
-    data = curd.model_to_dicts(schema=DictDataSchema, data=dict_data.items)
+    data = curd.model_to_dicts(schema=DictDataOutSchema, data=dict_data.items)
     return table_api(data=data, count=count)
 
 

applications/view/admin/monitor.py

@@ -5,11 +5,8 @@ from datetime import datetime
 import time
 import psutil
 from flask import Blueprint, render_template, jsonify
-from flask_marshmallow import Marshmallow
-
 from applications.common.utils.rights import authorize
 
-ma = Marshmallow()
 admin_monitor_bp = Blueprint('adminMonitor', __name__, url_prefix='/admin/monitor')
 
 

applications/view/admin/power.py

@@ -5,8 +5,8 @@ from applications.common.utils.http import success_api, fail_api
 from applications.common.utils.rights import authorize
 from applications.common.utils.validate import xss_escape
 from applications.extensions import db
-from applications.models import Power, Role
-from applications.schemas import PowerSchema2
+from applications.models import Power
+from applications.schemas import PowerOutSchema2
 
 admin_power = Blueprint('adminPower', __name__, url_prefix='/admin/power')
 
@@ -22,7 +22,7 @@ def index():
 def data():
     power = Power.query.all()
     res = {
-        "data": curd.model_to_dicts(schema=PowerSchema2, data=power)
+        "data": curd.model_to_dicts(schema=PowerOutSchema2, data=power)
     }
     return jsonify(res)
 
@@ -37,7 +37,7 @@ def add():
 @authorize("admin:power:main", log=True)
 def select_parent():
     power = Power.query.all()
-    res = curd.model_to_dicts(schema=PowerSchema2, data=power)
+    res = curd.model_to_dicts(schema=PowerOutSchema2, data=power)
     res.append({"powerId": 0, "powerName": "顶级权限", "parentId": -1})
     res = {
         "status": {"code": 200, "message": "默认"},
@@ -118,7 +118,7 @@ def update():
 def enable():
     _id = request.json.get('powerId')
     if id:
-        res = curd.enable_status(Power,_id)
+        res = curd.enable_status(Power, _id)
         if not res:
             return fail_api(msg="出错啦")
         return success_api(msg="启用成功")
@@ -131,7 +131,7 @@ def enable():
 def dis_enable():
     _id = request.json.get('powerId')
     if id:
-        res = curd.disable_status(Power,_id)
+        res = curd.disable_status(Power, _id)
         if not res:
             return fail_api(msg="出错啦")
         return success_api(msg="禁用成功")

applications/view/admin/role.py

@@ -7,7 +7,7 @@ from applications.common.utils.rights import authorize
 from applications.common.utils.validate import xss_escape
 from applications.extensions import db
 from applications.models import Role, Power, User
-from applications.schemas import RoleSchema, PowerSchema2
+from applications.schemas import RoleOutSchema, PowerOutSchema2
 
 admin_role = Blueprint('adminRole', __name__, url_prefix='/admin/role')
 
@@ -37,7 +37,7 @@ def table():
     role = Role.query.filter(mf.get_filter(Role)).layui_paginate()
     count = role.total
     # 返回api
-    return table_api(data=model_to_dicts(schema=RoleSchema, data=role.items), count=count)
+    return table_api(data=model_to_dicts(schema=RoleOutSchema, data=role.items), count=count)
 
 
 # 角色增加
@@ -86,7 +86,7 @@ def get_role_power(id):
     for cp in check_powers:
         check_powers_list.append(cp.id)
     powers = Power.query.all()
-    power_schema = PowerSchema2(many=True)  # 用已继承ma.ModelSchema类的自定制类生成序列化类
+    power_schema = PowerOutSchema2(many=True)  # 用已继承ma.ModelSchema类的自定制类生成序列化类
     output = power_schema.dump(powers)  # 生成可序列化对象
     for i in output:
         if int(i.get("powerId")) in check_powers_list:

applications/view/admin/user.py

@@ -11,7 +11,7 @@ from applications.common.utils.validate import xss_escape
 from applications.extensions import db
 from applications.models import Role
 from applications.models import User, AdminLog
-from applications.schemas import UserSchema
+from applications.schemas import UserOutSchema
 
 admin_user = Blueprint('adminUser', __name__, url_prefix='/admin/user')
 
@@ -44,7 +44,7 @@ def data():
     user = User.query.filter(mf.get_filter(model=User)).layui_paginate()
     count = user.total
     # 返回api
-    return table_api(data=model_to_dicts(schema=UserSchema, data=user.items), count=count)
+    return table_api(data=model_to_dicts(schema=UserOutSchema, data=user.items), count=count)
 
 
 # 用户增加

applications/view/department/__init__.py

@@ -1,5 +1,6 @@
 from flask import Blueprint, render_template, request, jsonify
 from marshmallow import INCLUDE
+from webargs.flaskparser import use_args
 
 from applications.common import curd
 from applications.common.utils.http import success_api, fail_api
@@ -7,7 +8,8 @@ from applications.common.utils.rights import authorize
 from applications.common.utils import validate
 from applications.extensions import db
 from applications.models import Dept, User
-from applications.schemas import DeptSchema
+from applications.schemas import DeptOutSchema
+from applications.schemas.admin_dept import DeptInSchema
 
 dept_bp = Blueprint('dept', __name__, url_prefix='/dept')
 
@@ -26,7 +28,7 @@ def main():
 @authorize("admin:dept:main", log=True)
 def data():
     dept = Dept.query.order_by(Dept.sort).all()
-    power_data = curd.model_to_dicts(schema=DeptSchema, data=dept)
+    power_data = curd.model_to_dicts(schema=DeptOutSchema, data=dept)
     res = {
         "data": power_data
     }
@@ -43,7 +45,7 @@ def add():
 @authorize("admin:dept:main", log=True)
 def tree():
     dept = Dept.query.order_by(Dept.sort).all()
-    power_data = curd.model_to_dicts(schema=DeptSchema, data=dept)
+    power_data = curd.model_to_dicts(schema=DeptOutSchema, data=dept)
     res = {
         "status": {"code": 200, "message": "默认"},
         "data": power_data
@@ -54,26 +56,17 @@ def tree():
 
 @dept_bp.post('/save')
 @authorize("admin:dept:add", log=True)
-def save():
-    req = request.json
-    validate.check_data(DeptSchema(unknown=INCLUDE), req)
-    address = validate.xss_escape(req.get("address"))
-    deptName = validate.xss_escape(req.get("deptName"))
-    email = validate.xss_escape(req.get("email"))
-    leader = validate.xss_escape(req.get("leader"))
-    parentId = validate.xss_escape(req.get("parentId"))
-    phone = validate.xss_escape(req.get("phone"))
-    sort = validate.xss_escape(req.get("sort"))
-    status = validate.xss_escape(req.get("status"))
+@use_args(DeptInSchema(), location="json", unknown=True)
+def save(args):
     dept = Dept(
-        parent_id=parentId,
-        dept_name=deptName,
-        sort=sort,
-        leader=leader,
-        phone=phone,
-        email=email,
-        status=status,
-        address=address
+        parent_id=args['parentId'],
+        dept_name=args['deptName'],
+        sort=args['sort'],
+        leader=args['leader'],
+        phone=args['phone'],
+        email=args['email'],
+        status=args['status'],
+        address=args['address']
     )
     r = db.session.add(dept)
     db.session.commit()

requirement/requirement-dev.txt

@@ -14,3 +14,4 @@ sqlparse==0.4.2
 captcha==0.3
 Pillow==8.2.0
 python-dotenv==0.19.1
+webargs==8.0.1