删除curd层

2021-07-12 13:04:16 +08:00
parent ca92d949dd
commit 3b3eecc1d1
18 changed files with 181 additions and 484 deletions
@@ -1,88 +0,0 @@
-from applications.common.utils.validate import xss_escape
-from applications.extensions import db
-from applications.models import Dept
-from applications.schemas import DeptSchema
-from applications.models import User
-from applications.common.curd import model_to_dicts
-
-
-def get_dept_dict():
-    dept = Dept.query.order_by(Dept.sort).all()
-    res = model_to_dicts(Schema=DeptSchema, model=dept)
-    return res
-
-
-def save_dept(req):
-    address = xss_escape(req.get("address"))
-    deptName = xss_escape(req.get("deptName"))
-    email = xss_escape(req.get("email"))
-    leader = xss_escape(req.get("leader"))
-    parentId = xss_escape(req.get("parentId"))
-    phone = xss_escape(req.get("phone"))
-    sort = xss_escape(req.get("sort"))
-    status = xss_escape(req.get("status"))
-    dept = Dept(
-        parent_id=parentId,
-        dept_name=deptName,
-        sort=sort,
-        leader=leader,
-        phone=phone,
-        email=email,
-        status=status,
-        address=address
-    )
-    r = db.session.add(dept)
-    db.session.commit()
-    return r
-
-
-def get_dept_by_id(id):
-    d = Dept.query.filter_by(id=id).first()
-    return d
-
-
-# 启动权限
-def enable_status(id):
-    enable = 1
-    d = Dept.query.filter_by(id=id).update({"status": enable})
-    if d:
-        db.session.commit()
-        return True
-    return False
-
-
-# 停用权限
-def disable_status(id):
-    enable = 0
-    d = Dept.query.filter_by(id=id).update({"status": enable})
-    if d:
-        db.session.commit()
-        return True
-    return False
-
-
-def update_dept(json):
-    id = json.get("deptId"),
-    data = {
-        "dept_name": xss_escape(json.get("deptName")),
-        "sort": xss_escape(json.get("sort")),
-        "leader": xss_escape(json.get("leader")),
-        "phone": xss_escape(json.get("phone")),
-        "email": xss_escape(json.get("email")),
-        "status": xss_escape(json.get("status")),
-        "address": xss_escape(json.get("address"))
-    }
-    d = Dept.query.filter_by(id=id).update(data)
-    if not d:
-        return False
-    db.session.commit()
-    return True
-
-
-def remove_dept(id):
-    d = Dept.query.filter_by(id=id).delete()
-    if not d:
-        return False
-    User.query.filter_by(dept_id=id).update({"dept_id": None})
-    db.session.commit()
-    return True
@@ -1,146 +0,0 @@
-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.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 = xss_escape(req_json.get("description"))
-    enable = xss_escape(req_json.get("enable"))
-    type_code = xss_escape(req_json.get("typeCode"))
-    type_name = xss_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 = xss_escape(req_json.get("id"))
-    description = xss_escape(req_json.get("description"))
-    enable = xss_escape(req_json.get("enable"))
-    type_code = xss_escape(req_json.get("typeCode"))
-    type_name = xss_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 = xss_escape(req_json.get("dataLabel"))
-    data_value = xss_escape(req_json.get("dataValue"))
-    enable = xss_escape(req_json.get("enable"))
-    remark = xss_escape(req_json.get("remark"))
-    type_code = xss_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": xss_escape(req_json.get("dataLabel")),
-        "data_value": xss_escape(req_json.get("dataValue")),
-        "enable": xss_escape(req_json.get("enable")),
-        "remark": xss_escape(req_json.get("remark")),
-        "type_code": xss_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
@@ -1,37 +0,0 @@
-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
-from applications.schemas import 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
@@ -1,113 +0,0 @@
-from applications.common.utils.validate import xss_escape
-from applications.extensions import db
-from applications.models.admin_power import Power
-from applications.schemas import PowerSchema2
-from applications.models import Role
-from applications.common.curd import model_to_dicts
-
-
-def get_power_dict():
-    power = Power.query.all()
-    res = model_to_dicts(Schema=PowerSchema2, model=power)
-    return res
-
-
-# 选择父节点
-def select_parent():
-    power = Power.query.all()
-    res = model_to_dicts(Schema=PowerSchema2, model=power)
-    res.append({"powerId": 0, "powerName": "顶级权限", "parentId": -1})
-    return res
-
-
-# 增加权限
-def save_power(req):
-    icon = xss_escape(req.get("icon"))
-    openType = xss_escape(req.get("openType"))
-    parentId = xss_escape(req.get("parentId"))
-    powerCode = xss_escape(req.get("powerCode"))
-    powerName = xss_escape(req.get("powerName"))
-    powerType = xss_escape(req.get("powerType"))
-    powerUrl = xss_escape(req.get("powerUrl"))
-    sort = xss_escape(req.get("sort"))
-    power = Power(
-        icon=icon,
-        open_type=openType,
-        parent_id=parentId,
-        code=powerCode,
-        name=powerName,
-        type=powerType,
-        url=powerUrl,
-        sort=sort,
-        enable=1
-    )
-    r = db.session.add(power)
-    db.session.commit()
-    return r
-
-
-# 根据id查询权限
-def get_power_by_id(id):
-    p = Power.query.filter_by(id=id).first()
-    return p
-
-
-# 更新权限
-def update_power(req_json):
-    id = req_json.get("powerId")
-    data = {
-        "icon": xss_escape(req_json.get("icon")),
-        "open_type": xss_escape(req_json.get("openType")),
-        "parent_id": xss_escape(req_json.get("parentId")),
-        "code": xss_escape(req_json.get("powerCode")),
-        "name": xss_escape(req_json.get("powerName")),
-        "type": xss_escape(req_json.get("powerType")),
-        "url": xss_escape(req_json.get("powerUrl")),
-        "sort": xss_escape(req_json.get("sort"))
-    }
-    # print(data)
-    power = Power.query.filter_by(id=id).update(data)
-    db.session.commit()
-    # print(power)
-    return power
-
-
-# 启动权限
-def enable_status(id):
-    enable = 1
-    user = Power.query.filter_by(id=id).update({"enable": enable})
-    if user:
-        db.session.commit()
-        return True
-    return False
-
-
-# 停用权限
-def disable_status(id):
-    enable = 0
-    user = Power.query.filter_by(id=id).update({"enable": enable})
-    if user:
-        db.session.commit()
-        return True
-    return False
-
-
-# 删除权限（目前没有判断父节点自动删除子节点）
-def remove_power(id):
-    power = Power.query.filter_by(id=id).first()
-    role_id_list = []
-    roles = power.role
-    for role in roles:
-        role_id_list.append(role.id)
-    roles = Role.query.filter(Role.id.in_(role_id_list)).all()
-    for p in roles:
-        power.role.remove(p)
-    r = Power.query.filter_by(id=id).delete()
-    db.session.commit()
-    return r
-
-
-# 批量删除权限
-def batch_remove(ids):
-    for id in ids:
-        remove_power(id)
@@ -1,10 +1,15 @@
-from applications.extensions import db
+from applications.extensions import db, ma


-def model_to_dicts(Schema, model):
+def model_to_dicts(schema: ma.Schema, data):
+    """
+    :param schema: schema类
+    :param model: sqlalchemy查询结果
+    :return: 返回单个查询结果
+    """
    # 如果是分页器返回，需要传入model.items
-    common_schema = Schema(many=True)  # 用已继承ma.ModelSchema类的自定制类生成序列化类
-    output = common_schema.dump(model)  # 生成可序列化对象
+    common_schema = schema(many=True)  # 用已继承ma.ModelSchema类的自定制类生成序列化类
+    output = common_schema.dump(data)  # 生成可序列化对象
    return output


@@ -1,17 +0,0 @@
-#
-MYSQL_HOST = '127.0.0.1'
-MYSQL_PORT = '3306'
-MYSQL_DATABASE = 'PearAdminFlask'
-MYSQL_USERNAME = 'root'
-MYSQL_PASSWORD = 'root'
-
-MYSQL_DB_URI = "mysql+pymysql://{username}:{password}@{host}:{port}/{db}?charset=utf8".format(username=USERNAME,
-                                                                                        password=PASSWORD,
-                                                                                        host=HOST,
-                                                                                        port=PORT,
-                                                                                        db=DATABASE)
-
-SQLALCHEMY_DATABASE_URI = MYSQL_DB_URI
-SQLALCHEMY_TRACK_MODIFICATIONS = True
-SQLALCHEMY_ECHO = False
-SQLALCHEMY_POOL_RECYCLE = 8
@@ -2,3 +2,39 @@ from flask_uploads import UploadSet, IMAGES


 photos = UploadSet('photos', IMAGES)
+
+import os
+from flask import current_app
+from sqlalchemy import desc
+from applications.extensions import db
+from applications.models import Photo
+from applications.schemas import 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, data=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