用户注册登录实现后,下一步编写查询用户列表接口,使用flask_restful的marshal方法来生成字段数据,加上之前编写的login_required
from flask import Blueprint,jsonify from flask_restful import fields, marshal from auth import login_required from models.user import User userbp = Blueprint('user',__name__,url_prefix="/user") user_fields = { 'id': fields.Integer, 'username': fields.String, 'nickname': fields.String, 'email': fields.String, 'group_id': fields.Integer, '_create_time': fields.DateTime(dt_format='iso8601'), 'update_time': fields.DateTime(dt_format='iso8601'), } @userbp.route("/list",methods=['GET']) @login_required def list(): users = User.getall() return jsonify(code=201,msg="操作成功",data=marshal(users, user_fields))
请求时header中不加入Authorization字段就会验证失败
在header中添加Authorization,值为Bearer+空格+login返回数据的access_token,成功获取list接口数据
用户的增删改查都是一样的逻辑来写
from flask import Blueprint, jsonify from flask_restful import fields, marshal, reqparse from werkzeug.security import generate_password_hash from auth import login_required from models.user import User userbp = Blueprint('user',__name__,url_prefix="/user") user_fields = { 'id': fields.Integer, 'username': fields.String, 'nickname': fields.String, 'email': fields.String, 'group_id': fields.Integer, '_create_time': fields.DateTime(dt_format='iso8601'), 'update_time': fields.DateTime(dt_format='iso8601'), } @userbp.route("/list",methods=['GET']) @login_required def list(): users = User.getall() return jsonify(code=201,msg="操作成功",data=marshal(users, user_fields)) @userbp.route("/<uid>",methods=['GET']) @login_required def info(uid): user = User.getone(uid) return jsonify({ "code": 0, "msg": "success", "data": marshal(user, user_fields) }) @userbp.route("/add",methods=['POST']) @login_required def add(): parser = reqparse.RequestParser() parser.add_argument("username", type=str, required=True, help="username is required") parser.add_argument("nickname", type=str, required=True, help="nickname is required") parser.add_argument('password', required=True, type=str, help='password is required') parser.add_argument('email', required=False, type=str) parser.add_argument('group_id', required=False, type=int, default=2) args = parser.parse_args() if User.exist(username=args['username']): return jsonify(code=400,msg="用户名已存在") args.update({'password': generate_password_hash(args['password'], salt_length=8)}) User.add(args) return jsonify(code=201,msg="添加成功") @userbp.route("/edit",methods=['POST']) @login_required def edit(): parser = reqparse.RequestParser() parser.add_argument("id", required=True, type=int, help="id is required") parser.add_argument("username", type=str, required=True, help="username is required") parser.add_argument("nickname", type=str, required=True, help="nickname is required") parser.add_argument('email', required=False, type=str) parser.add_argument('group_id', required=False, type=int) args = parser.parse_args() if not User.exist(id=args['id']): return jsonify(code=201,msg="用户不存在") User.edit(args) return jsonify(code=201,msg="修改成功") @userbp.route("/delete",methods=['GET']) @login_required def delete(): parser = reqparse.RequestParser() parser.add_argument("id", required=True, type=int, location='args',help="id is required") args = parser.parse_args() if not User.exist(id=args['id']): return jsonify(code=201,msg="用户不存在") User.remove(args['id']) return jsonify(code=201, msg="删除成功")
自定义权限校验
之后为接口添加分组的权限校验,根目录添加文件router.py,编写路由装饰器
routes ={} def route_meta(auth,endname):
# endname必须是{Blueprint注册的名字.方法名} def wrapper(func): routes.setdefault(f"{endname}", auth) return func return wrapper
在auth.py添加校验路由的装饰器
def group_required(fn): @wraps(fn) def wrapper(*args, **kwargs): if not current_user.isadmin: auth = routes.get(request.endpoint) groupid = current_user.group_id authed = Auth.exist(group_id=groupid,auth=auth,endpoint=request.endpoint) if not authed: return jsonify(code=10000,msg="权限不够,请联系超级管理员获得权限") return fn(*args,**kwargs) else: return fn(*args,**kwargs) return wrapper
在add接口方法加上这两个装饰器,route_meta第二个参数必须是{Blueprint注册的名字.方法名}
@route_meta("新增用户", "user.add") @userview.route("/add",methods=['POST']) @login_required @group_required def add():
......
再次请求add接口会提示没有权限
{ "code": 10000, "msg": "权限不够,请联系超级管理员获得权限" } 之后就需要为用户添加权限,权限关联到相关分组,所以添加修改分组接口@groupbp.route("/edit",methods=['POST']) @login_required def edit(): parser = reqparse.RequestParser() parser.add_argument("id", required=True, type=int, help="id is required") parser.add_argument("name", required=True, type=str, help="name is required") parser.add_argument("info", required=False, type=str) args = parser.parse_args() args['auths'] = request.json['auths'] sql = select(Group).where(Group.name==args['name'],Group.id!=args['id']) if db.session.execute(sql).first(): return jsonify(code=10000,msg="分组名已存在") Group.edit(args) # 需要重写edit return jsonify(code=201,msg="操作成功")
由于修改分组需要添加、删除auth表的数据,所以要在models/group.py文件增加自己的edit方法
from models.auth import Auth
class Group:
......
@classmethod def edit(cls, data): group = cls.getone(data['id']) group.name = data['name'] group.info = data['info'] for auth in group.auths: Auth.remove(auth.id) for item in data.get('auths'): authobj = Auth.add(dict(endpoint=item,auth=routes[item])) group.auths.append(authobj) db.session.commit()
通过group/edit接口可以为admin分组增加权限
![](/i/l/?n=22&i=blog/788784/202210/788784-20221016012709588-258488411.png)
上面接口成功的话就为admin增加了add和edit这两个方法的权限,再添加和修改用户就能成功了
{ "code": 201, "msg": "添加成功" } 其它用户登录操作add、edit则会提示权限不够,当为delete方法添加group_required装饰器后,admin删除用户也会提示权限不够 至此flask用户管理及权限验证的后台接口基本后端框架就完成了,还有一些细节需要自己去改进,然后就是选择前端框架编写前端页面。标签:parser,args,flask,fields,入门教程,改查,required,add,id From: https://www.cnblogs.com/zerotest/p/16797023.html