我们还可以像上一节那样，从复杂对象中创建token。在本例中，我们可以获取一个令牌，并且每次访问受保护的端点时，都会自动使用该令牌来加载复杂对象(例如SQLAlchemy实例)。这是通过user_loader_callback_loader()装饰器完成的。通过使用get_current_user()函数，或者直接使用current_user LocalProxy，可以在受保护的端点中访问结果对象。需要注意的一件事是，如果您在user_loader_callback_loader()中访问数据库，那么每次调用时都会产生数据库查找的成本，而不管您是否需要来自数据库的额外数据。在大多数情况下，这可能不是什么值得担心的事情，但请注意，如果它处理高流量，它可能会减慢您的应用程序。
下面是这个功能的一个例子:
from flask import Flask, jsonify, request
from flask_jwt_extended import (
JWTManager, jwt_required, create_access_token, current_user
)
app = Flask(__name__)
app.config['JWT_SECRET_KEY'] = 'super-secret' # Change this!
jwt = JWTManager(app)
# A demo user object that we will use in this example.
class UserObject:
def __init__(self, username, roles):
self.username = username
self.roles = roles
# An example store of users. In production, this would likely
# be a sqlalchemy instance or something similar.
users_to_roles = {
'foo': ['admin'],
'bar': ['peasant'],
'baz': ['peasant']
}
# This function is called whenever a protected endpoint is accessed,
# and must return an object based on the tokens identity.
# This is called after the token is verified, so you can use
# get_jwt_claims() in here if desired. Note that this needs to
# return None if the user could not be loaded for any reason,
# such as not being found in the underlying data store
@jwt.user_loader_callback_loader
def user_loader_callback(identity):
if identity not in users_to_roles:
return None
return UserObject(
username=identity,
roles=users_to_roles[identity]
)
# You can override the error returned to the user if the
# user_loader_callback returns None. If you don't override
# this, # it will return a 401 status code with the JSON:
# {"msg": "Error loading the user <identity>"}.
# You can use # get_jwt_claims() here too if desired
@jwt.user_loader_error_loader
def custom_user_loader_error(identity):
ret = {
"msg": "User {} not found".format(identity)
}
return jsonify(ret), 404
# Create a token for any user, so this can be tested out
@app.route('/login', methods=['POST'])
def login():
username = request.get_json().get('username', None)
access_token = create_access_token(identity=username)
ret = {'access_token': access_token}
return jsonify(ret), 200
# If the user_loader_callback returns None, this method will
# not be run, even if the access token is valid. You can
# access the loaded user via the ``current_user``` LocalProxy,
# or with the ```get_current_user()``` method
@app.route('/admin-only', methods=['GET'])
@jwt_required
def protected():
if 'admin' not in current_user.roles:
return jsonify({"msg": "Forbidden"}), 403
else:
return jsonify({"msg": "don't forget to drink your ovaltine"})
if __name__ == '__main__':
app.run()
