增加多条件查询构造器，删除部分curd层
This commit is contained in:
mkg
@@ -1,150 +0,0 @@
|
||||
from sqlalchemy import and_
|
||||
from applications.common.utils.validate import xss_escape
|
||||
from applications.extensions import db
|
||||
from applications.models import Role
|
||||
from applications.schemas import RoleSchema
|
||||
from applications.models.admin_power import Power
|
||||
from applications.schemas import PowerSchema2
|
||||
from applications.models import User
|
||||
|
||||
# 获取角色对象
|
||||
from applications.common.curd import model_to_dicts
|
||||
|
||||
|
||||
def get_role_data(page, limit, filters):
|
||||
role = Role.query.filter(and_(*[getattr(Role, k).like(v) for k, v in filters.items()])).paginate(page=page,
|
||||
per_page=limit,
|
||||
error_out=False)
|
||||
count = Role.query.count()
|
||||
return role, count
|
||||
|
||||
|
||||
# 获取角色dict
|
||||
def get_role_data_dict(page, limit, filters):
|
||||
role, count = get_role_data(page, limit, filters)
|
||||
data = model_to_dicts(Schema=RoleSchema, model=role.items)
|
||||
return data, count
|
||||
|
||||
|
||||
# 增加角色
|
||||
def add_role(req):
|
||||
details = xss_escape(req.get("details"))
|
||||
enable = xss_escape(req.get("enable"))
|
||||
roleCode = xss_escape(req.get("roleCode"))
|
||||
roleName = xss_escape(req.get("roleName"))
|
||||
sort = xss_escape(req.get("sort"))
|
||||
role = Role(
|
||||
details=details,
|
||||
enable=enable,
|
||||
code=roleCode,
|
||||
name=roleName,
|
||||
sort=sort
|
||||
)
|
||||
db.session.add(role)
|
||||
db.session.commit()
|
||||
|
||||
|
||||
# 通过id获取角色
|
||||
def get_role_by_id(id):
|
||||
r = Role.query.filter_by(id=id).first()
|
||||
return r
|
||||
|
||||
|
||||
# 更新角色
|
||||
def update_role(req_json):
|
||||
id = req_json.get("roleId")
|
||||
data = {
|
||||
"code": xss_escape(req_json.get("roleCode")),
|
||||
"name": xss_escape(req_json.get("roleName")),
|
||||
"sort": xss_escape(req_json.get("sort")),
|
||||
"enable": xss_escape(req_json.get("enable")),
|
||||
"details": xss_escape(req_json.get("details"))
|
||||
}
|
||||
role = Role.query.filter_by(id=id).update(data)
|
||||
db.session.commit()
|
||||
return role
|
||||
|
||||
|
||||
# 获取角色的权限
|
||||
def get_role_power(id):
|
||||
role = Role.query.filter_by(id=id).first()
|
||||
check_powers = role.power
|
||||
check_powers_list = []
|
||||
for cp in check_powers:
|
||||
check_powers_list.append(cp.id)
|
||||
powers = Power.query.all()
|
||||
power_schema = PowerSchema2(many=True) # 用已继承ma.ModelSchema类的自定制类生成序列化类
|
||||
output = power_schema.dump(powers) # 生成可序列化对象
|
||||
for i in output:
|
||||
if int(i.get("powerId")) in check_powers_list:
|
||||
i["checkArr"] = "1"
|
||||
else:
|
||||
i["checkArr"] = "0"
|
||||
return output
|
||||
|
||||
|
||||
# 更新角色权限
|
||||
def update_role_power(id, power_list):
|
||||
role = Role.query.filter_by(id=id).first()
|
||||
power_id_list = []
|
||||
for p in role.power:
|
||||
power_id_list.append(p.id)
|
||||
# print(p.id)
|
||||
# print(power_id_list)
|
||||
powers = Power.query.filter(Power.id.in_(power_id_list)).all()
|
||||
for p in powers:
|
||||
role.power.remove(p)
|
||||
powers = Power.query.filter(Power.id.in_(power_list)).all()
|
||||
for p in powers:
|
||||
role.power.append(p)
|
||||
db.session.commit()
|
||||
|
||||
|
||||
# 启动角色
|
||||
def enable_status(id):
|
||||
enable = 1
|
||||
role = Role.query.filter_by(id=id).update({"enable": enable})
|
||||
if role:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# 停用角色
|
||||
def disable_status(id):
|
||||
enable = 0
|
||||
role = Role.query.filter_by(id=id).update({"enable": enable})
|
||||
if role:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# 删除角色
|
||||
def remove_role(id):
|
||||
role = Role.query.filter_by(id=id).first()
|
||||
# 删除该角色的权限
|
||||
power_id_list = []
|
||||
for p in role.power:
|
||||
power_id_list.append(p.id)
|
||||
|
||||
powers = Power.query.filter(Power.id.in_(power_id_list)).all()
|
||||
for p in powers:
|
||||
role.power.remove(p)
|
||||
user_id_list = []
|
||||
for u in role.user:
|
||||
user_id_list.append(u.id)
|
||||
users = User.query.filter(User.id.in_(user_id_list)).all()
|
||||
for u in users:
|
||||
role.user.remove(u)
|
||||
r = Role.query.filter_by(id=id).delete()
|
||||
db.session.commit()
|
||||
return r
|
||||
|
||||
|
||||
# 批量删除
|
||||
def batch_remove(ids):
|
||||
# role = Role.query.filter(Role.id.in_(ids)).delete(synchronize_session=False)
|
||||
# db.session.commit()
|
||||
for id in ids:
|
||||
remove_role(id)
|
||||
@@ -1,160 +0,0 @@
|
||||
from flask import jsonify
|
||||
from flask_login import current_user
|
||||
from sqlalchemy import and_, desc
|
||||
from applications.extensions import db
|
||||
from applications.models import User
|
||||
from applications.schemas import UserSchema
|
||||
from applications.models import Role
|
||||
from applications.models import AdminLog
|
||||
|
||||
# 获取用户的 sqlalchemy 对象分页器
|
||||
from applications.common.curd import model_to_dicts
|
||||
|
||||
|
||||
def get_user_data(page, limit, filters, deptId):
|
||||
if deptId:
|
||||
user = User.query.filter_by(dept_id=deptId).filter(
|
||||
and_(*[getattr(User, k).like(v) for k, v in filters.items()])).paginate(page=page,
|
||||
per_page=limit,
|
||||
error_out=False)
|
||||
else:
|
||||
user = User.query.filter(and_(*[getattr(User, k).like(v) for k, v in filters.items()])).paginate(page=page,
|
||||
per_page=limit,
|
||||
error_out=False)
|
||||
count = User.query.count()
|
||||
return user, count
|
||||
|
||||
|
||||
# 获取用户的dict数据分页器
|
||||
def get_user_data_dict(page, limit, filters, deptId):
|
||||
user, count = get_user_data(page, limit, filters, deptId)
|
||||
data = model_to_dicts(Schema=UserSchema, model=user.items)
|
||||
return data, count
|
||||
|
||||
|
||||
# 通过名称获取用户
|
||||
def get_user_by_name(username):
|
||||
return User.query.filter_by(username=username).first()
|
||||
|
||||
|
||||
# 获取当前用户日志
|
||||
def get_current_user_logs():
|
||||
log = AdminLog.query.filter_by(url='/admin/login').filter_by(uid=current_user.id).order_by(
|
||||
desc(AdminLog.create_time)).limit(10)
|
||||
return log
|
||||
|
||||
|
||||
# 判断用户是否存在
|
||||
def is_user_exists(username):
|
||||
res = User.query.filter_by(username=username).count()
|
||||
return bool(res)
|
||||
|
||||
|
||||
# 增加用户
|
||||
def add_user(username, realName, password):
|
||||
user = User(username=username, realname=realName)
|
||||
user.set_password(password)
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
return user.id
|
||||
|
||||
|
||||
# 增加用户角色
|
||||
def add_user_role(id, roles_list):
|
||||
user = User.query.filter_by(id=id).first()
|
||||
roles = Role.query.filter(Role.id.in_(roles_list)).all()
|
||||
for r in roles:
|
||||
user.role.append(r)
|
||||
db.session.commit()
|
||||
|
||||
|
||||
# 更新用户头像
|
||||
def update_avatar(url):
|
||||
r = User.query.filter_by(id=current_user.id).update({"avatar": url})
|
||||
db.session.commit()
|
||||
return r
|
||||
|
||||
|
||||
# 更新用户信息
|
||||
def update_user(id, username, realname, deptId):
|
||||
user = User.query.filter_by(id=id).update({'username': username, 'realname': realname, 'dept_id': deptId})
|
||||
db.session.commit()
|
||||
return user
|
||||
|
||||
|
||||
# 更新当前用户信息
|
||||
def update_current_user_info(req_json):
|
||||
r = User.query.filter_by(id=current_user.id).update(
|
||||
{"realname": req_json.get("realName"), "remark": req_json.get("details")})
|
||||
db.session.commit()
|
||||
return r
|
||||
|
||||
|
||||
# 修改当前用户密码
|
||||
def edit_password(res_json):
|
||||
if res_json.get("newPassword") == '':
|
||||
return jsonify(success=False, msg="新密码不得为空")
|
||||
if res_json.get("newPassword") != res_json.get("confirmPassword"):
|
||||
return jsonify(success=False, msg="俩次密码不一样")
|
||||
user = current_user
|
||||
is_right = user.validate_password(res_json.get("oldPassword"))
|
||||
if not is_right:
|
||||
return jsonify(success=False, msg="旧密码错误")
|
||||
user.set_password(res_json.get("newPassword"))
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
return jsonify(success=True, msg="更改成功")
|
||||
|
||||
|
||||
# 删除用户
|
||||
def delete_by_id(id):
|
||||
user = User.query.filter_by(id=id).first()
|
||||
roles_id = []
|
||||
for role in user.role:
|
||||
roles_id.append(role.id)
|
||||
roles = Role.query.filter(Role.id.in_(roles_id)).all()
|
||||
for r in roles:
|
||||
user.role.remove(r)
|
||||
res = User.query.filter_by(id=id).delete()
|
||||
db.session.commit()
|
||||
return res
|
||||
|
||||
|
||||
# 启用用户
|
||||
def enable_status(id):
|
||||
enable = 1
|
||||
user = User.query.filter_by(id=id).update({"enable": enable})
|
||||
if user:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# 停用用户
|
||||
def disable_status(id):
|
||||
enable = 0
|
||||
user = User.query.filter_by(id=id).update({"enable": enable})
|
||||
if user:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# 批量删除
|
||||
def batch_remove(ids):
|
||||
for id in ids:
|
||||
delete_by_id(id)
|
||||
|
||||
|
||||
def update_user_role(id, roles_list):
|
||||
user = User.query.filter_by(id=id).first()
|
||||
roles_id = []
|
||||
for role in user.role:
|
||||
roles_id.append(role.id)
|
||||
roles = Role.query.filter(Role.id.in_(roles_id)).all()
|
||||
for r in roles:
|
||||
user.role.remove(r)
|
||||
roles = Role.query.filter(Role.id.in_(roles_list)).all()
|
||||
for r in roles:
|
||||
user.role.append(r)
|
||||
db.session.commit()
|
||||
@@ -1,5 +1,37 @@
|
||||
from applications.extensions import db
|
||||
|
||||
|
||||
def model_to_dicts(Schema, model):
|
||||
# 如果是分页器返回，需要传入model.items
|
||||
common_schema = Schema(many=True) # 用已继承ma.ModelSchema类的自定制类生成序列化类
|
||||
output = common_schema.dump(model) # 生成可序列化对象
|
||||
return output
|
||||
|
||||
|
||||
def get_one_by_id(model, id):
|
||||
"""
|
||||
:param model: 模型类
|
||||
:param id: id
|
||||
:return: 返回单个查询结果
|
||||
"""
|
||||
return model.query.filter_by(id=id).first()
|
||||
|
||||
|
||||
# 启动状态
|
||||
def enable_status(model, id):
|
||||
enable = 1
|
||||
role = model.query.filter_by(id=id).update({"enable": enable})
|
||||
if role:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# 停用状态
|
||||
def disable_status(model, id):
|
||||
enable = 0
|
||||
role = model.query.filter_by(id=id).update({"enable": enable})
|
||||
if role:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
from sqlalchemy import and_
|
||||
|
||||
from applications.extensions import db
|
||||
|
||||
|
||||
class ModelFilter:
|
||||
"""
|
||||
orm多参数构造器
|
||||
"""
|
||||
filter_field = {}
|
||||
filter_list = []
|
||||
|
||||
type_exact = "exact"
|
||||
type_vague = "vague"
|
||||
|
||||
def __init__(self):
|
||||
self.filter_field = {}
|
||||
self.filter_list = []
|
||||
|
||||
def exact(self, field_name, value):
|
||||
"""
|
||||
准确查询字段
|
||||
:param field_name: 模型字段名称
|
||||
:param value: 值
|
||||
"""
|
||||
if value and value != '':
|
||||
self.filter_field[field_name] = {"data": value, "type": self.type_exact}
|
||||
|
||||
def vague(self, field_name, value: str):
|
||||
"""
|
||||
模糊查询字段
|
||||
:param field_name: 模型字段名称
|
||||
:param value: 值
|
||||
"""
|
||||
if value and value != '':
|
||||
self.filter_field[field_name] = {"data": ('%' + value + '%'), "type": self.type_vague}
|
||||
|
||||
def get_filter(self, model: db.Model):
|
||||
"""
|
||||
获取过滤条件
|
||||
:param model: 模型字段名称
|
||||
"""
|
||||
for k, v in self.filter_field.items():
|
||||
if v.get("type") == self.type_vague:
|
||||
self.filter_list.append(getattr(model, k).like(v.get("data")))
|
||||
if v.get("type") == self.type_exact:
|
||||
self.filter_list.append(getattr(model, k) == v.get("data"))
|
||||
return and_(*self.filter_list)
|
||||
Reference in New Issue
Block a user