使用flask-restful实现数据校验与序列化2
This commit is contained in:
zhengxinonly
@@ -1,7 +1,6 @@
|
||||
from flask import Flask
|
||||
|
||||
from applications.view.admin.logs import admin_log
|
||||
from applications.view.admin.dict import admin_dict
|
||||
from applications.view.admin.index import admin_bp
|
||||
from applications.view.admin.file import file_bp
|
||||
from applications.view.rights.role import role_bp
|
||||
@@ -16,4 +15,3 @@ def register_admin_views(app: Flask):
|
||||
app.register_blueprint(admin_monitor_bp)
|
||||
app.register_blueprint(admin_log)
|
||||
app.register_blueprint(role_bp)
|
||||
app.register_blueprint(admin_dict)
|
||||
|
||||
@@ -1,178 +0,0 @@
|
||||
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.view.admin import dict_curd
|
||||
|
||||
admin_dict = Blueprint('adminDict', __name__, url_prefix='/admin/dict')
|
||||
|
||||
|
||||
# 数据字典
|
||||
@admin_dict.get('/')
|
||||
@authorize("admin:dict:main", log=True)
|
||||
def main():
|
||||
return render_template('admin/dict/main.html')
|
||||
|
||||
|
||||
@admin_dict.get('/dictType/data')
|
||||
@authorize("admin:dict:main", log=True)
|
||||
def dict_type_data():
|
||||
page = request.args.get('page', type=int)
|
||||
limit = request.args.get('limit', type=int)
|
||||
type_name = request.args.get('typeName', type=str)
|
||||
data, count = dict_curd.get_dict_type(page=page, limit=limit, type_name=type_name)
|
||||
return table_api(data=data, count=count)
|
||||
|
||||
|
||||
@admin_dict.get('/dictType/add')
|
||||
@authorize("admin:dict:add", log=True)
|
||||
def dict_type_add():
|
||||
return render_template('admin/dict/add.html')
|
||||
|
||||
|
||||
@admin_dict.post('/dictType/save')
|
||||
@authorize("admin:dict:add", log=True)
|
||||
def dict_type_save():
|
||||
req_json = request.json
|
||||
res = dict_curd.save_dict_type(req_json=req_json)
|
||||
if res is None:
|
||||
return fail_api(msg="增加失败")
|
||||
return success_api(msg="增加成功")
|
||||
|
||||
|
||||
# 编辑字典类型
|
||||
@admin_dict.get('/dictType/edit')
|
||||
@authorize("admin:dict:edit", log=True)
|
||||
def dict_type_edit():
|
||||
_id = request.args.get('dictTypeId', type=int)
|
||||
dict_type = DictType.query.filter_by(id=_id).first()
|
||||
return render_template('admin/dict/edit.html', dict_type=dict_type)
|
||||
|
||||
|
||||
# 编辑字典类型
|
||||
@admin_dict.put('/dictType/update')
|
||||
@authorize("admin:dict:edit", log=True)
|
||||
def dict_type_update():
|
||||
req_json = request.json
|
||||
dict_curd.update_dict_type(req_json)
|
||||
return success_api(msg="更新成功")
|
||||
|
||||
|
||||
# 启用字典
|
||||
@admin_dict.put('/dictType/enable')
|
||||
@authorize("admin:dict:edit", log=True)
|
||||
def dict_type_enable():
|
||||
_id = request.json.get('id')
|
||||
if id:
|
||||
res = dict_curd.enable_dict_type_status(_id)
|
||||
if not res:
|
||||
return fail_api(msg="出错啦")
|
||||
return success_api("启动成功")
|
||||
return fail_api(msg="数据错误")
|
||||
|
||||
|
||||
# 禁用字典
|
||||
@admin_dict.put('/dictType/disable')
|
||||
@authorize("admin:dict:edit", log=True)
|
||||
def dict_type_dis_enable():
|
||||
_id = request.json.get('id')
|
||||
if id:
|
||||
res = dict_curd.disable_dict_type_status(_id)
|
||||
if not res:
|
||||
return fail_api(msg="出错啦")
|
||||
return success_api("禁用成功")
|
||||
return fail_api(msg="数据错误")
|
||||
|
||||
|
||||
# 删除字典类型
|
||||
@admin_dict.delete('/dictType/remove/<int:_id>')
|
||||
@authorize("admin:dict:remove", log=True)
|
||||
def dict_type_delete(_id):
|
||||
res = dict_curd.delete_type_by_id(_id)
|
||||
if not res:
|
||||
return fail_api(msg="删除失败")
|
||||
return success_api(msg="删除成功")
|
||||
|
||||
|
||||
@admin_dict.get('/dictData/data')
|
||||
@authorize("admin:dict:main", log=True)
|
||||
def dict_code_data():
|
||||
page = request.args.get('page', type=int)
|
||||
limit = request.args.get('limit', type=int)
|
||||
type_code = request.args.get('typeCode', type=str)
|
||||
data, count = dict_curd.get_dict_data(page=page, limit=limit, type_code=type_code)
|
||||
return table_api(data=data,count=count)
|
||||
|
||||
|
||||
# 增加字典数据
|
||||
@admin_dict.get('/dictData/add')
|
||||
@authorize("admin:dict:add", log=True)
|
||||
def dict_data_add():
|
||||
type_code = request.args.get('typeCode', type=str)
|
||||
return render_template('admin/dict/data/add.html', type_code=type_code)
|
||||
|
||||
|
||||
# 增加字典数据
|
||||
@admin_dict.get('/dictData/save')
|
||||
@authorize("admin:dict:add", log=True)
|
||||
def dict_data_save():
|
||||
req_json = request.json
|
||||
res = dict_curd.save_dict_data(req_json=req_json)
|
||||
if not res:
|
||||
return jsonify(success=False, msg="增加失败")
|
||||
return jsonify(success=True, msg="增加成功")
|
||||
|
||||
|
||||
# 编辑字典数据
|
||||
@admin_dict.get('/dictData/edit')
|
||||
@authorize("admin:dict:edit", log=True)
|
||||
def dict_data_edit():
|
||||
_id = request.args.get('dataId', type=str)
|
||||
dict_data = DictData.query.filter_by(id=_id).first()
|
||||
return render_template('admin/dict/data/edit.html', dict_data=dict_data)
|
||||
|
||||
|
||||
# 编辑字典数据
|
||||
@admin_dict.put('/dictData/update')
|
||||
@authorize("admin:dict:edit", log=True)
|
||||
def dict_data_update():
|
||||
req_json = request.json
|
||||
dict_curd.update_dict_data(req_json)
|
||||
return success_api(msg="更新成功")
|
||||
|
||||
|
||||
# 启用字典数据
|
||||
@admin_dict.put('/dictData/enable')
|
||||
@authorize("admin:dict:edit", log=True)
|
||||
def dict_data_enable():
|
||||
_id = request.json.get('dataId')
|
||||
if _id:
|
||||
res = dict_curd.enable_dict_data_status(_id)
|
||||
if not res:
|
||||
return fail_api(msg="出错啦")
|
||||
return success_api(msg="启动成功")
|
||||
return fail_api(msg="数据错误")
|
||||
|
||||
|
||||
# 禁用字典数据
|
||||
@admin_dict.put('/dictData/disable')
|
||||
@authorize("admin:dict:edit", log=True)
|
||||
def dict_data_disenable():
|
||||
_id = request.json.get('dataId')
|
||||
if _id:
|
||||
res = dict_curd.disable_dict_data_status(_id)
|
||||
if not res:
|
||||
return fail_api(msg="出错啦")
|
||||
return success_api(msg="禁用成功")
|
||||
return fail_api(msg="数据错误")
|
||||
|
||||
|
||||
# 删除字典类型
|
||||
@admin_dict.delete('dictData/remove/<int:id>')
|
||||
@authorize("admin:dict:remove", log=True)
|
||||
def dict_data_delete(id):
|
||||
res = dict_curd.delete_data_by_id(id)
|
||||
if not res:
|
||||
return fail_api(msg="删除失败")
|
||||
return success_api(msg="删除成功")
|
||||
@@ -1,145 +0,0 @@
|
||||
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
|
||||
@@ -5,7 +5,7 @@ rights_bp = Blueprint('rights', __name__, url_prefix='/rights')
|
||||
rights_api = Api(rights_bp)
|
||||
|
||||
from . import routes
|
||||
from . import view
|
||||
from . import right
|
||||
|
||||
|
||||
def register_rights_view(app: Flask):
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
from flask import render_template, request, jsonify, make_response
|
||||
from flask_restful import Resource, reqparse
|
||||
|
||||
from . import rights_bp, rights_api, right_curd
|
||||
from ...common.utils.http import success_api, fail_api
|
||||
from ...common.utils.rights import authorize
|
||||
from ...extensions import db
|
||||
from ...models import Power
|
||||
|
||||
|
||||
@rights_bp.get('/')
|
||||
@authorize("admin:power:main", log=True)
|
||||
def index():
|
||||
return render_template('admin/rights/main.html')
|
||||
|
||||
|
||||
@rights_bp.get('/data')
|
||||
@authorize("admin:power:main", log=True)
|
||||
def data():
|
||||
power_data = right_curd.get_power_dict()
|
||||
res = {
|
||||
"data": power_data
|
||||
}
|
||||
return jsonify(res)
|
||||
|
||||
|
||||
@rights_api.resource('/power')
|
||||
class AddRight(Resource):
|
||||
@authorize("admin:power:add", log=True)
|
||||
def get(self):
|
||||
"""获取增加视图"""
|
||||
return make_response(render_template('admin/rights/add.html'))
|
||||
|
||||
@authorize("admin:power:add", log=True)
|
||||
def post(self):
|
||||
req = request.json
|
||||
right_curd.save_power(req)
|
||||
return success_api(msg="成功")
|
||||
|
||||
|
||||
@rights_bp.get('/selectParent')
|
||||
@authorize("admin:power:main", log=True)
|
||||
def select_parent():
|
||||
"""获取选择父节点"""
|
||||
power_data = right_curd.select_parent()
|
||||
res = {
|
||||
"status": {"code": 200, "message": "默认"},
|
||||
"data": power_data
|
||||
|
||||
}
|
||||
return jsonify(res)
|
||||
|
||||
|
||||
@rights_api.resource('/power/<int:power_id>')
|
||||
class RightsURD(Resource):
|
||||
@authorize("admin:power:edit", log=True)
|
||||
def get(self, power_id):
|
||||
power = right_curd.get_power_by_id(power_id)
|
||||
icon = str(power.icon).split()
|
||||
if len(icon) == 2:
|
||||
icon = icon[1]
|
||||
else:
|
||||
icon = None
|
||||
return make_response(render_template('admin/rights/edit.html', power=power, icon=icon))
|
||||
|
||||
@authorize("admin:power:remove", log=True)
|
||||
def delete(self, power_id):
|
||||
r = right_curd.remove_power(power_id)
|
||||
if r:
|
||||
return success_api(msg="删除成功")
|
||||
else:
|
||||
return fail_api(msg="删除失败")
|
||||
|
||||
@authorize("admin:power:edit", log=True)
|
||||
def put(self, power_id):
|
||||
parser = reqparse.RequestParser(bundle_errors=True)
|
||||
parser.add_argument('icon', type=str)
|
||||
parser.add_argument('openType', type=str, dest='open_type')
|
||||
parser.add_argument('parentId', type=str, dest='parent_id')
|
||||
parser.add_argument('powerCode', type=str, dest='power_code')
|
||||
parser.add_argument('powerName', type=str, dest='power_name')
|
||||
parser.add_argument('powerType', type=str, dest='power_type')
|
||||
parser.add_argument('powerUrl', type=str, dest='power_url')
|
||||
parser.add_argument('sort', type=int, dest='sort')
|
||||
|
||||
res = parser.parse_args()
|
||||
data = {
|
||||
"icon": res.icon,
|
||||
"open_type": res.open_type,
|
||||
"parent_id": res.parent_id,
|
||||
"code": res.power_code,
|
||||
"name": res.power_name,
|
||||
"type": res.power_type,
|
||||
"url": res.power_url,
|
||||
"sort": res.sort
|
||||
}
|
||||
power = Power.query.filter_by(id=power_id).update(data)
|
||||
db.session.commit()
|
||||
|
||||
if not power:
|
||||
return fail_api(msg="更新权限失败")
|
||||
return success_api(msg="更新权限成功")
|
||||
|
||||
|
||||
# 启用权限
|
||||
@rights_bp.put('/enable')
|
||||
@authorize("admin:power:edit", log=True)
|
||||
def enable():
|
||||
parser = reqparse.RequestParser(bundle_errors=True)
|
||||
parser.add_argument('powerId', dest='power_id', type=int, required=True)
|
||||
parser.add_argument('operate', type=int, required=True, choices=[0, 1])
|
||||
|
||||
res = parser.parse_args()
|
||||
|
||||
power = Power.query.filter_by(id=res.power_id).update({"enable": res.operate})
|
||||
if power:
|
||||
db.session.commit()
|
||||
message = "启用成功" if res.operate else '禁用成功'
|
||||
return success_api(msg=message)
|
||||
else:
|
||||
return fail_api(msg="出错啦")
|
||||
|
||||
|
||||
# 批量删除
|
||||
@rights_bp.delete('/batchRemove')
|
||||
@authorize("admin:power:remove", log=True)
|
||||
def batch_remove():
|
||||
ids = request.form.getlist('ids[]')
|
||||
right_curd.batch_remove(ids)
|
||||
return success_api(msg="批量删除成功")
|
||||
@@ -1,114 +0,0 @@
|
||||
from flask import render_template, request, jsonify, make_response
|
||||
from flask_restful import Resource
|
||||
|
||||
from . import rights_bp, rights_api, rights_curd
|
||||
from ...common.utils.http import success_api, fail_api
|
||||
from ...common.utils.rights import authorize
|
||||
|
||||
|
||||
@rights_bp.get('/')
|
||||
@authorize("admin:power:main", log=True)
|
||||
def index():
|
||||
return render_template('admin/power/main.html')
|
||||
|
||||
|
||||
@rights_bp.get('/data')
|
||||
@authorize("admin:power:main", log=True)
|
||||
def data():
|
||||
power_data = rights_curd.get_power_dict()
|
||||
res = {
|
||||
"data": power_data
|
||||
}
|
||||
return jsonify(res)
|
||||
|
||||
|
||||
@rights_bp.get('/add')
|
||||
@authorize("admin:power:add", log=True)
|
||||
def add():
|
||||
"""获取增加视图"""
|
||||
return render_template('admin/power/add.html')
|
||||
|
||||
|
||||
@rights_bp.get('/selectParent')
|
||||
@authorize("admin:power:main", log=True)
|
||||
def select_parent():
|
||||
"""获取选择父节点"""
|
||||
power_data = rights_curd.select_parent()
|
||||
res = {
|
||||
"status": {"code": 200, "message": "默认"},
|
||||
"data": power_data
|
||||
|
||||
}
|
||||
return jsonify(res)
|
||||
|
||||
|
||||
# 增加
|
||||
@rights_bp.post('/save')
|
||||
@authorize("admin:power:add", log=True)
|
||||
def save():
|
||||
req = request.json
|
||||
rights_curd.save_power(req)
|
||||
return success_api(msg="成功")
|
||||
|
||||
|
||||
@rights_api.resource('/power/<int:power_id>')
|
||||
class RightsURD(Resource):
|
||||
@authorize("admin:power:edit", log=True)
|
||||
def get(self, power_id):
|
||||
power = rights_curd.get_power_by_id(power_id)
|
||||
icon = str(power.icon).split()
|
||||
if len(icon) == 2:
|
||||
icon = icon[1]
|
||||
else:
|
||||
icon = None
|
||||
return make_response(render_template('admin/power/edit.html', power=power, icon=icon))
|
||||
|
||||
@authorize("admin:power:remove", log=True)
|
||||
def delete(self, power_id):
|
||||
r = rights_curd.remove_power(power_id)
|
||||
if r:
|
||||
return success_api(msg="删除成功")
|
||||
else:
|
||||
return fail_api(msg="删除失败")
|
||||
|
||||
@authorize("admin:power:edit", log=True)
|
||||
def put(self, power_id):
|
||||
res = rights_curd.update_power(request.json)
|
||||
if not res:
|
||||
return fail_api(msg="更新权限失败")
|
||||
return success_api(msg="更新权限成功")
|
||||
|
||||
|
||||
# 启用权限
|
||||
@rights_bp.put('/enable')
|
||||
@authorize("admin:power:edit", log=True)
|
||||
def enable():
|
||||
power_id = request.json.get('powerId')
|
||||
if id:
|
||||
res = rights_curd.enable_status(power_id)
|
||||
if not res:
|
||||
return fail_api(msg="出错啦")
|
||||
return success_api(msg="启用成功")
|
||||
return fail_api(msg="数据错误")
|
||||
|
||||
|
||||
# 禁用权限
|
||||
@rights_bp.put('/disable')
|
||||
@authorize("admin:power:edit", log=True)
|
||||
def dis_enable():
|
||||
_id = request.json.get('powerId')
|
||||
if _id:
|
||||
res = rights_curd.disable_status(_id)
|
||||
if not res:
|
||||
return fail_api(msg="出错啦")
|
||||
return success_api(msg="禁用成功")
|
||||
return fail_api(msg="数据错误")
|
||||
|
||||
|
||||
# 批量删除
|
||||
@rights_bp.delete('/batchRemove')
|
||||
@authorize("admin:power:remove", log=True)
|
||||
def batch_remove():
|
||||
ids = request.form.getlist('ids[]')
|
||||
rights_curd.batch_remove(ids)
|
||||
return success_api(msg="批量删除成功")
|
||||
Reference in New Issue
Block a user