Flask-JWT-Extended basic usage sample code

main.py

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': '[email protected]', 'password': 'taro'},
    {'id': 2, 'username': 'Hanako', 'email': '[email protected]', '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()

test.py

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': '[email protected]',
            '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': '[email protected]', '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': '[email protected]',
            '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': '[email protected]',
            '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()