flask_jwt 用户认证

Posted 2021-02-01 alextd

from config.config import jwt
from flask import jsonify
from flask_jwt_extended import get_jwt_identity, jwt_required, get_raw_jwt,jwt_optional
from common.ReturnMessage import returnNoneMsg
from flask import request
from functools import wraps
from models.User import User
import types
from sqlalchemy import or_

#必选登录用户
def user_require(fn):
    @jwt_required
    @wraps(fn)
    def _deco():
        if type(request.json) != types.DictType:
            resultDict = returnNoneMsg("failed!")
            return jsonify(resultDict)
        userId = get_jwt_identity()
        if not userId:
            resultDict = returnNoneMsg("not find userId!")
            return jsonify(resultDict)
        userTable = User.query.filter(User.id==userId,
                                          or_(User.is_lock == 1,User.is_lock == 2)).first()
        postJwt = get_raw_jwt()
        postCounselorToken = postJwt.get("jti")
        if userTable:
            userToken = userTable.access_token
            if userToken == postCounselorToken:
                return fn()
            else:
                resultDict = returnNoneMsg("this user not exit!")
        else:
            resultDict = returnNoneMsg("this user not exit!")
        return jsonify(resultDict)
    return _deco

#用户既可登陆也不登录
def user_or_require(fn):
    @jwt_optional
    @wraps(fn)
    def _deco():
        if type(request.json) != types.DictType:
            resultDict = returnNoneMsg("failed!")
            return jsonify(resultDict)
        userId = get_jwt_identity()
        if userId:
            userTable = User.query.filter(User.id == userId,
                                          or_(User.is_lock == 1,User.is_lock == 2)).first()
            postJwt = get_raw_jwt()
            postCounselorToken = postJwt.get("jti")
            if userTable:
                userToken = userTable.access_token
                if userToken == postCounselorToken:
                    return fn()
                else:
                    resultDict = returnNoneMsg("the token not mach with sql!")
            else:
                resultDict = returnNoneMsg("this user not exit!")
            return jsonify(resultDict)
        else :
            return fn()
    return _deco


@jwt.invalid_token_loader
@jwt.unauthorized_loader
def my_unauthorized_loader(identy):
    dictInfo = "The has no token: %s" % identy
    resultDict = returnNoneMsg(dictInfo)
    return jsonify(resultDict)

@jwt.expired_token_loader
def my_expired_token_callback():
    returnInfo = returnNoneMsg("the token expired!")
    return jsonify(returnInfo)

# 访问受保护端口 会调用此函数
@jwt.user_loader_callback_loader

# 认证失败 自动调用此函数
@jwt.user_loader_error_loader