我正在使用烧瓶 - jwt扩展来保护我的烧瓶API。登录后,在受保护的路线(附加用户)中,我致电get_jwt_identity(),但它没有返回,所以我无法获得身份。
@flask_app.route('/<api_version>/login', methods=['POST'])
def login(api_version):
print(f'login', request.form)
response = None
try:
username = request.form['username']
password = request.form['password']
if not username:
return jsonify({"msg": "Missing username parameter"}), 400
if not password:
return jsonify({"msg": "Missing password parameter"}), 400
user = User.get_with_password(username, password)
if (not user):
e1 = Unauthorized(
'Invalid username or password. Please try again.')
e1.status = 401
raise e1
""" flask_login.login_user(user, remember=True) """
access_token = create_access_token(identity=username)
response = json.dumps({"token": access_token}, cls=CustomJSONEncoder)
except Exception as e:
errMsg = f'Error Logging in user {username if username else ""}: {e}'
status = e.status if hasattr(e, 'status') else 500
print(f'{errMsg}')
traceback.print_exc()
return Response(
json.dumps({"message": errMsg, "status": status, "stack": traceback.format_exc() }), status=status, mimetype='application/json')
resp = Response(response, status=200, mimetype='application/json')
return resp
@flask_app.route('/<api_version>/add-user', methods=['POST'])
@jwt_required
def add_user(api_version):
print(f'add-user', request)
response = None
username = None
password = None
allow_admin = None
try:
data = request.get_json()
print(f'add-user data', data)
if 'username' in data:
username = data['username']
else:
return jsonify({"msg": "Missing username parameter"}), 400
if 'password' in data:
password = data['password']
else:
return jsonify({"msg": "Missing password parameter"}), 400
if 'allow_admin' in data:
allow_admin = data['allow_admin']
""" user = User.get_with_password(username, password)"""
user = get_jwt_identity()
print('user',user)
if (not user):
e1 = Unauthorized(
'Invalid username or password. Please try again.')
e1.status = 401
raise e1
response = json.dumps({"user": user}, cls=CustomJSONEncoder)
except Exception as e:
errMsg = f'Error Adding User {username}: {e}'
status = e.status if hasattr(e, 'status') else 500
print(f'{errMsg}')
traceback.print_exc()
return Response(
json.dumps({"message": errMsg, "status": status, "stack": traceback.format_exc() }), status=status, mimetype='application/json')
resp = Response(response, status=200, mimetype='application/json')
return resp
user.py
class User():
@classmethod
def get_with_password(cls, username, password):
print(f'User get_with_password {username} with password')
user_db = account.get_account(username)
print(f'User returned from DB: {user_db}')
user = User()
if not user_db or not len(user_db) or (not 'password' in user_db):
return None
user.username = username
user.id = username
if bcrypt.check_password_hash(user_db['password'], password):
user.role = user_db['role']
#user.is_authenticated = True
print(
f'loginUser returning {vars(user)} ')
return user
return None
我也面临同样的问题,在我的情况下,我忘了添加@jwt_required
@jwt_required
def get(self):
uid = get_jwt_identity()
print("self.uid",uid)
在您的代码@jwt_required中,在那里,检查可能是token已过期
我遇到了同样的问题,并通过以下步骤解决了它
step1:检查是否已包装了应用程序(flask_app)如下。
from flask_jwt_extended import JWTManager
jwt = JWTManager(flask_app)
step2:检查应用程序中是否有JWT_BLACKLIST_ENABLED。
如果jwt_blacklist_enabled = false或jwt_blacklist_enabled不存在,请在上面添加以下方法。
@jwt.user_claims_loader
def add_claims_to_access_token(identity):
return {
'identity': identity
}
如果jwt_blacklist_enabled = true,请在登录方法上方添加以下内容。
@jwt.token_in_blacklist_loader
def check_if_token_in_blacklist(decrypted_token):
jti = decrypted_token['jti']
return jti in blacklist