|
| 1 | +from flask import Flask, jsonify, request |
| 2 | +from flask_jwt_extended import ( |
| 3 | + JWTManager, jwt_required, create_access_token, current_user |
| 4 | +) |
| 5 | + |
| 6 | +app = Flask(__name__) |
| 7 | +app.secret_key = 'super-secret' # Change this! |
| 8 | +jwt = JWTManager(app) |
| 9 | + |
| 10 | + |
| 11 | +# A user object that we will load our tokens |
| 12 | +class UserObject: |
| 13 | + def __init__(self, username, roles): |
| 14 | + self.username = username |
| 15 | + self.roles = roles |
| 16 | + |
| 17 | +# An example store of users. In production, this would likely |
| 18 | +# be a sqlalchemy instance or something similiar |
| 19 | +users_to_roles = { |
| 20 | + 'foo': ['admin'], |
| 21 | + 'bar': ['peasant'], |
| 22 | + 'baz': ['peasant'] |
| 23 | +} |
| 24 | + |
| 25 | + |
| 26 | +# This function is called whenever a protected endpoint is accessed. |
| 27 | +# This should return a complex object based on the token identity. |
| 28 | +# This is called after the token is verified, so you can use |
| 29 | +# get_jwt_claims() in here if desired. Note that this needs to |
| 30 | +# return None if the user could not be loaded for any reason, |
| 31 | +# such as not being found in the underlying data store |
| 32 | +@jwt.user_loader_callback_loader |
| 33 | +def user_loader_callback(identity): |
| 34 | + if identity not in users_to_roles: |
| 35 | + return None |
| 36 | + |
| 37 | + return UserObject( |
| 38 | + username=identity, |
| 39 | + roles=users_to_roles[identity] |
| 40 | + ) |
| 41 | + |
| 42 | + |
| 43 | +# You can override the error returned to the user if the |
| 44 | +# user_loader_callback returns None. By default, if you don't |
| 45 | +# override this, it will return a 401 status code with the json: |
| 46 | +# {'msg': "Error loading the user <identity>"}. You can use |
| 47 | +# get_jwt_claims() here too if desired |
| 48 | +@jwt.user_loader_error_loader |
| 49 | +def custom_user_loader_error(identity): |
| 50 | + return jsonify({"msg": "User not found"}), 404 |
| 51 | + |
| 52 | + |
| 53 | +# Create a token for any user, so this can be tested out |
| 54 | +@app.route('/login', methods=['POST']) |
| 55 | +def login(): |
| 56 | + username = request.json.get('username', None) |
| 57 | + access_token = create_access_token(identity=username) |
| 58 | + ret = {'access_token': access_token} |
| 59 | + return jsonify(ret), 200 |
| 60 | + |
| 61 | + |
| 62 | +# If the user_loader_callback returns None, this method will |
| 63 | +# not get hit, even if the access token is valid. You can |
| 64 | +# access the loaded user via the ``current_user``` LocalProxy, |
| 65 | +# or with the ```get_current_user()``` method |
| 66 | +@app.route('/admin-only', methods=['GET']) |
| 67 | +@jwt_required |
| 68 | +def protected(): |
| 69 | + if 'admin' not in current_user.roles: |
| 70 | + return jsonify({"msg": "Forbidden"}), 403 |
| 71 | + return jsonify({"secret_msg": "don't forget to drink your ovaltine"}) |
| 72 | + |
| 73 | +if __name__ == '__main__': |
| 74 | + app.run() |
0 commit comments