Created
May 4, 2020 08:25
-
-
Save KentaYamada/ba28685c2facb5f59c128d322dd5da7e to your computer and use it in GitHub Desktop.
Flask-JWT-Extended basic usage sample code
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from flask import jsonify, request, Flask | |
| from flask_jwt_extended import ( | |
| jwt_required, | |
| create_access_token, | |
| JWTManager | |
| ) | |
| HTTP_OK = 200 | |
| HTTP_BAD_REQUEST = 400 | |
| HTTP_UNAUTHORIZED = 401 | |
| users = [ | |
| {'id': 1, 'username': 'Taro', 'email': 'taro@email.com', 'password': 'taro'}, | |
| {'id': 2, 'username': 'Hanako', 'email': 'hanako@email.com', 'password': 'hanako'} | |
| ] | |
| def jwt_unauthorized_loader_handler(reason): | |
| return jsonify({'message': 'Unauthorized'}), HTTP_UNAUTHORIZED | |
| app = Flask(__name__) | |
| app.url_map.strict_slashes = False | |
| # Set up Flask-JWT-Extended extention | |
| app.config['JWT_SECRET_KEY'] = 'secret-key' | |
| jwt = JWTManager(app) | |
| jwt.unauthorized_loader(jwt_unauthorized_loader_handler) | |
| @app.route('/api/login', methods=['POST']) | |
| def login(): | |
| if not request.is_json: | |
| body = {'message': 'Missing JSON in request'} | |
| return jsonify(body), HTTP_BAD_REQUEST | |
| request_body = request.get_json() | |
| if request_body is None: | |
| body = {'message': 'Request body is empty'} | |
| return jsonify(body), HTTP_BAD_REQUEST | |
| whitelist = {'email', 'password'} | |
| if not request_body.keys() <= whitelist: | |
| body = {'message': 'Missing email or password in request'} | |
| return jsonify(body), HTTP_BAD_REQUEST | |
| auth_user = None | |
| for user in users: | |
| isMatchEmail = user['email'] == request_body['email'] | |
| isMatchPassword = user['password'] == request_body['password'] | |
| if isMatchEmail and isMatchPassword: | |
| auth_user = user | |
| break | |
| if auth_user is None: | |
| body = {'message': 'Login failure. Bad email or password'} | |
| return jsonify(body), HTTP_UNAUTHORIZED | |
| token = create_access_token(identity=auth_user['username']) | |
| body = {'message': 'Login succeeded', 'token': token} | |
| return jsonify(body), HTTP_OK | |
| @app.route('/api/users', methods=['GET']) | |
| @jwt_required | |
| def index(): | |
| return jsonify({'users': users}), HTTP_OK | |
| if __name__ == '__main__': | |
| app.run() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import json | |
| import unittest | |
| from main import app | |
| class TestFlaskJwtExtendedServer(unittest.TestCase): | |
| CONTENT_TYPE = 'application/json' | |
| def test_login_succeeded(self): | |
| req = json.dumps({ | |
| 'email': 'taro@email.com', | |
| 'password': 'taro' | |
| }) | |
| res = self.api.post( | |
| '/api/login', | |
| content_type=self.CONTENT_TYPE, | |
| data=req | |
| ) | |
| body = res.get_json() | |
| self.assertEqual(200, res.status_code) | |
| self.assertEqual(body['message'], 'Login succeeded') | |
| def test_login_failure_when_send_invalid_content_type(self): | |
| res = self.api.post( | |
| '/api/login', | |
| content_type='text/html' | |
| ) | |
| body = res.get_json() | |
| self.assertEqual(400, res.status_code) | |
| self.assertEqual(body['message'], 'Missing JSON in request') | |
| def test_login_failure_when_send_empty_request(self): | |
| res = self.api.post( | |
| '/api/login', | |
| content_type=self.CONTENT_TYPE, | |
| data=json.dumps(None) | |
| ) | |
| body = res.get_json() | |
| self.assertEqual(400, res.status_code) | |
| self.assertEqual(body['message'], 'Request body is empty') | |
| def test_login_failure_when_invalid_post_data(self): | |
| data = [ | |
| {'hoge': 'hoge', 'fuga': 'fuga'}, | |
| {'email': 'taro@email.com', 'fuga': 'fuga'}, | |
| {'hoge': 'hoge', 'password': 'taro'} | |
| ] | |
| for d in data: | |
| res = self.api.post( | |
| '/api/login', | |
| content_type=self.CONTENT_TYPE, | |
| data=json.dumps(d) | |
| ) | |
| body = res.get_json() | |
| self.assertEqual(400, res.status_code) | |
| self.assertEqual( | |
| body['message'], | |
| 'Missing email or password in request' | |
| ) | |
| def test_login_failure_when_wrong_data(self): | |
| req = json.dumps({ | |
| 'email': 'hoge@email.com', | |
| 'password': 'taro' | |
| }) | |
| res = self.api.post( | |
| '/api/login', | |
| content_type=self.CONTENT_TYPE, | |
| data=req | |
| ) | |
| body = res.get_json() | |
| self.assertEqual(401, res.status_code) | |
| self.assertEqual( | |
| body['message'], | |
| 'Login failure. Bad email or password' | |
| ) | |
| def test_api_users_succeeded(self): | |
| # get token | |
| req = json.dumps({ | |
| 'email': 'taro@email.com', | |
| 'password': 'taro' | |
| }) | |
| res = self.api.post( | |
| '/api/login', | |
| content_type=self.CONTENT_TYPE, | |
| data=req | |
| ) | |
| body = res.get_json() | |
| headers = { | |
| 'Authorization': 'Bearer {}'.format(body['token']) | |
| } | |
| res2 = self.api.get( | |
| '/api/users', | |
| content_type=self.CONTENT_TYPE, | |
| headers=headers | |
| ) | |
| self.assertEqual(200, res2.status_code) | |
| def test_api_users_failure_when_no_auth_header(self): | |
| res = self.api.get( | |
| '/api/users', | |
| content_type=self.CONTENT_TYPE, | |
| ) | |
| body = res.get_json() | |
| self.assertEqual(401, res.status_code) | |
| self.assertEqual(body['message'], 'Unauthorized') | |
| @classmethod | |
| def setUpClass(cls): | |
| cls.api = app.test_client() | |
| @classmethod | |
| def tearDownClass(cls): | |
| cls.api = None | |
| if __name__ == '__main__': | |
| unittest.main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment