使用flask-restful实现数据校验与序列化

applications/view/admin/dict.py

@@ -3,7 +3,7 @@ from flask import Blueprint, render_template, request, jsonify
 from applications.common.utils.http import table_api, success_api, fail_api
 from applications.common.utils.rights import authorize
 from applications.models import DictType, DictData
-from applications.common.admin import dict_curd
+from applications.view.admin import dict_curd
 
 admin_dict = Blueprint('adminDict', __name__, url_prefix='/admin/dict')
 

applications/view/admin/dict_curd.py

@@ -0,0 +1,145 @@
+from flask import escape
+from applications.extensions import db
+from applications.models import DictType, DictData, DictTypeSchema, DictDataSchema
+from applications.common.curd import model_to_dicts
+
+
+def get_dict(typecode: str):
+    dict_list = []
+    if DictType.query.filter_by(type_code=typecode, enable=1).first():
+        dicts = DictData.query.filter_by(type_code=typecode, enable=1).all()
+        for d in dicts:
+            dict_dict = {"key": d.data_label, "value": d.data_value}
+            dict_list.append(dict_dict)
+    else:
+        return None
+    return dict_list
+
+
+def get_dict_type(page, limit, type_name):
+    dict_all = DictType.query
+    if type_name:
+        dict_all = dict_all.filter(DictType.type_name.like('%' + type_name + '%'))
+    dict_all = dict_all.paginate(page=page,
+                                 per_page=limit,
+                                 error_out=False)
+    count = DictType.query.count()
+    data = model_to_dicts(Schema=DictTypeSchema, model=dict_all.items)
+    return data, count
+
+
+def get_dict_data(page, limit, type_code):
+    dict_all = DictData.query.filter_by(type_code=type_code).paginate(page=page,
+                                                                      per_page=limit,
+                                                                      error_out=False)
+    count = DictType.query.count()
+    dict_dict = model_to_dicts(Schema=DictDataSchema, model=dict_all.items)
+    return dict_dict, count
+
+
+# 增加 dict type
+def save_dict_type(req_json):
+    description = escape(req_json.get("description"))
+    enable = escape(req_json.get("enable"))
+    type_code = escape(req_json.get("typeCode"))
+    type_name = escape(req_json.get("typeName"))
+    d = DictType(type_name=type_name, type_code=type_code, enable=enable, description=description)
+    db.session.add(d)
+    db.session.commit()
+    return d.id
+
+
+# 编辑字典类型
+def update_dict_type(req_json):
+    id = escape(req_json.get("id"))
+    description = escape(req_json.get("description"))
+    enable = escape(req_json.get("enable"))
+    type_code = escape(req_json.get("typeCode"))
+    type_name = escape(req_json.get("typeName"))
+    DictType.query.filter_by(id=id).update({
+        "description": description,
+        "enable": enable,
+        "type_code": type_code,
+        "type_name": type_name
+    })
+    db.session.commit()
+    return
+
+
+def enable_dict_type_status(id):
+    enable = 1
+    res = DictType.query.filter_by(id=id).update({"enable": enable})
+    if res:
+        db.session.commit()
+        return True
+    return False
+
+
+def disable_dict_type_status(id):
+    enable = 0
+    res = DictType.query.filter_by(id=id).update({"enable": enable})
+    if res:
+        db.session.commit()
+        return True
+    return False
+
+
+# 删除字典类型
+def delete_type_by_id(id):
+    type_code = DictType.query.filter_by(id=id).first().type_code
+    DictData.query.filter_by(type_code=type_code).delete()
+    res = DictType.query.filter_by(id=id).delete()
+    db.session.commit()
+    return res
+
+
+# 增加dictdata
+def save_dict_data(req_json):
+    data_label = escape(req_json.get("dataLabel"))
+    data_value = escape(req_json.get("dataValue"))
+    enable = escape(req_json.get("enable"))
+    remark = escape(req_json.get("remark"))
+    type_code = escape(req_json.get("typeCode"))
+    d = DictData(data_label=data_label, data_value=data_value, enable=enable, remark=remark, type_code=type_code)
+    db.session.add(d)
+    db.session.commit()
+    return d.id
+
+
+# 编辑字典数据
+def update_dict_data(req_json):
+    id = req_json.get("dataId")
+    DictData.query.filter_by(id=id).update({
+        "data_label": escape(req_json.get("dataLabel")),
+        "data_value": escape(req_json.get("dataValue")),
+        "enable": escape(req_json.get("enable")),
+        "remark": escape(req_json.get("remark")),
+        "type_code": escape(req_json.get("typeCode"))
+    })
+    db.session.commit()
+    return
+
+
+def enable_dict_data_status(id):
+    enable = 1
+    res = DictData.query.filter_by(id=id).update({"enable": enable})
+    if res:
+        db.session.commit()
+        return True
+    return False
+
+
+def disable_dict_data_status(id):
+    enable = 0
+    res = DictData.query.filter_by(id=id).update({"enable": enable})
+    if res:
+        db.session.commit()
+        return True
+    return False
+
+
+# 删除dictdata
+def delete_data_by_id(id):
+    res = DictData.query.filter_by(id=id).delete()
+    db.session.commit()
+    return res

applications/view/admin/file.py

@@ -1,13 +1,14 @@
 import os
-from flask import Blueprint, request, render_template, jsonify, current_app
-
+from flask import Blueprint, request, render_template, jsonify, current_app, make_response
+from flask_restful import Api, Resource
 from applications.common.utils.http import fail_api, success_api, table_api
 from applications.common.utils.rights import authorize
 from applications.extensions import db
 from applications.models import Photo
-from applications.common.admin import file_curd
+from applications.view.admin import file_curd
 
 file_bp = Blueprint('file', __name__, url_prefix='/file')
+file_api = Api(file_bp)
 
 
 #  图片管理
@@ -27,42 +28,36 @@ def table():
     return table_api(data=data, count=count)
 
 
-#   上传
-@file_bp.get('/upload')
-@authorize("admin:file:add", log=True)
-def upload():
-    return render_template('admin/file/photo_add.html')
+@file_api.resource('/upload')
+class Upload(Resource):
+    @authorize("admin:file:add", log=True)
+    def get(self):
+        return make_response(render_template('admin/file/photo_add.html'))
 
+    @authorize("admin:file:add", log=True)
+    def post(self):
+        if 'file' in request.files:
+            photo = request.files['file']
+            mime = request.files['file'].content_type
+            file_url = file_curd.upload_one(photo=photo, mime=mime)
+            res = {
+                "msg": "上传成功",
+                "code": 0,
+                "success": True,
+                "data":
+                    {"src": file_url}
+            }
+            return jsonify(res)
+        return fail_api()
 
-#   上传接口
-@file_bp.post('/upload')
-@authorize("admin:file:add", log=True)
-def upload_api():
-    if 'file' in request.files:
-        photo = request.files['file']
-        mime = request.files['file'].content_type
-        file_url = file_curd.upload_one(photo=photo, mime=mime)
-        res = {
-            "msg": "上传成功",
-            "code": 0,
-            "success": True,
-            "data":
-                {"src": file_url}
-        }
-        return jsonify(res)
-    return fail_api()
-
-
-#    图片删除
-@file_bp.route('/delete', methods=['GET', 'POST'])
-@authorize("admin:file:delete", log=True)
-def delete():
-    _id = request.form.get('id')
-    res = file_curd.delete_photo_by_id(_id)
-    if res:
-        return success_api(msg="删除成功")
-    else:
-        return fail_api(msg="删除失败")
+    @authorize("admin:file:delete", log=True)
+    def delete(self):
+        _id = request.form.get('id')
+        res = file_curd.delete_photo_by_id(_id)
+        if res:
+            return success_api(msg="删除成功")
+        else:
+            return fail_api(msg="删除失败")
 
 
 # 图片批量删除

applications/view/admin/file_curd.py

@@ -0,0 +1,36 @@
+import os
+from flask import current_app
+from sqlalchemy import desc
+
+from applications.common.utils.upload import photos
+from applications.extensions import db
+from applications.models import Photo, PhotoSchema
+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, model=photo.items)
+    return data, count
+
+
+def upload_one(photo, mime):
+    filename = photos.save(photo)
+    file_url = photos.url(filename)
+
+    upload_url = current_app.config.get("UPLOADED_PHOTOS_DEST")
+    size = os.path.getsize(upload_url + '/' + filename)
+    photo = Photo(name=filename, href=file_url, mime=mime, size=size)
+    db.session.add(photo)
+    db.session.commit()
+    return file_url
+
+
+def delete_photo_by_id(_id):
+    photo_name = Photo.query.filter_by(id=_id).first().name
+    photo = Photo.query.filter_by(id=_id).delete()
+    db.session.commit()
+    upload_url = current_app.config.get("UPLOADED_PHOTOS_DEST")
+    os.remove(upload_url + '/' + photo_name)
+    return photo