Flask + flask-jwt 实现基于Json Web Token的用户认证授权

上一篇文章《Flask + PyJWT 实现基于Json Web Token的用户认证授权》讲了Flask整合PyJWT实现Json Web Token用户认证授权的例子。而专门针对Flask,有一个flask-jwt库也可以实现基于Json Web Token的用户认证授权,这篇文章就来讲下Flask整合flask-jwt实现Json Web Token用户认证授权。

 

一、需求

示例需求与前一篇文章类似,可参考《Flask + PyJWT 实现基于Json Web Token的用户认证授权》

 

二、程序目录结构

程序目录结构与前一篇文章一样

 

三、程序实现

1、程序构建及相关文件

数据迁移配置文件:

from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand
from run import app
from app import db

app.config.from_object('app.config')

db.init_app(app)

migrate = Migrate(app, db)
manager = Manager(app)
manager.add_command('db', MigrateCommand)

if __name__ == '__main__':
    manager.run()

运行入口文件:

from app import create_app

app = create_app('app.config')

if __name__ == '__main__':
    app.run(host=app.config['HOST'],
            port=app.config['PORT'],
            debug=app.config['DEBUG'])

程序初始化文件:

from flask import Flask, request
from flask_sqlalchemy import SQLAlchemy
from flask_jwt import JWT

db = SQLAlchemy()

def create_app(config_filename):
    app = Flask(__name__)
    app.config.from_object(config_filename)

    @app.after_request
    def after_request(response):
        response.headers.add('Access-Control-Allow-Origin', '*')
        if request.method == 'OPTIONS':
            response.headers['Access-Control-Allow-Methods'] = 'DELETE, GET, POST, PUT'
            headers = request.headers.get('Access-Control-Request-Headers')
            if headers:
                response.headers['Access-Control-Allow-Headers'] = headers
        return response

    from app.auth.auths import Auth
    auth = Auth()
    jwt = JWT(app, auth.authenticate, auth.identity)

    from app.users.model import db
    db.init_app(app)

    from app.users.api import init_api
    init_api(app)

    return app

与上篇文章使用PyJWT不同的是,这里导入了flask-jwt和认证类,并且直接初始化jwt认证模块。

from app.auth.auths import Auth
auth = Auth()
jwt = JWT(app, auth.authenticate, auth.identity)

这样,就可以在全局使用认证,具体使用和认证类方法我们往后看。

配置文件:

DB_USER = 'root'
DB_PASSWORD = ''
DB_HOST = 'localhost'
DB_DB = 'flask-pyjwt-auth'

DEBUG = True
PORT = 3333
HOST = "192.168.1.141"
SECRET_KEY = "my blog"

JWT_AUTH_URL_RULE = '/login'

SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_DATABASE_URI = 'mysql://' + DB_USER + ':' + DB_PASSWORD + '@' + DB_HOST + '/' + DB_DB

配置文件中,也有一点点跟原来使用PyJWT不一样。这里定义了配置变量JWT_AUTH_URL_RULE,与PyJWT不同,flask-jwt默认给我们定义了认证路由,其路由终端(Endpoint)是”/auth”,即使用http://192.168.1.141:3333/auth来得到认证token,而这里的JWT_AUTH_URL_RULE就是用来改变这个认证路由终端的,这样设置后,就可以使用我们熟悉的http://192.168.1.141:3333/login来获取认证token

Flask-jwt提供了十几个配置项,用来配置JWT的一些实现细节。如:

公共文件:

def trueReturn(data, msg):
    return {
        "status": True,
        "data": data,
        "msg": msg
    }


def falseReturn(data, msg):
    return {
        "status": False,
        "data": data,
        "msg": msg
    }

2、用户模块

模块入口(空)


用户模型:

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.exc import SQLAlchemyError
from werkzeug.security import generate_password_hash, check_password_hash

from app import db

class Users(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(250),  unique=True, nullable=False)
    username = db.Column(db.String(250),  unique=True, nullable=False)
    password = db.Column(db.String(250))
    login_time = db.Column(db.Integer)

    def __init__(self, username, password, email):
        self.username = username
        self.password = password
        self.email = email

    def __str__(self):
        return "Users(id='%s')" % self.id

    def set_password(self, password):
        return generate_password_hash(password)

    def check_password(self, hash, password):
        return check_password_hash(hash, password)

    def get(self, id):
        return self.query.filter_by(id=id).first()

    def add(self, user):
        db.session.add(user)
        return session_commit()

    def update(self):
        return session_commit()

    def delete(self, id):
        self.query.filter_by(id=id).delete()
        return session_commit()


def session_commit():
    try:
        db.session.commit()
    except SQLAlchemyError as e:
        db.session.rollback()
        reason = str(e)
        return reason

用户相关接口实现:

from flask import jsonify, request
from app.users.model import Users
from flask_jwt import jwt_required, current_identity
from .. import common

def init_api(app):
    @app.route('/register', methods=['POST'])
    def register():
        """
        用户注册
        :return: json
        """
        email = request.form.get('email')
        username = request.form.get('username')
        password = request.form.get('password')
        user = Users(email=email, username=username, password=Users.set_password(Users, password))
        result = Users.add(Users, user)
        if user.id:
            returnUser = {
                'id': user.id,
                'username': user.username,
                'email': user.email,
                'login_time': user.login_time
            }
            return jsonify(common.trueReturn(returnUser, "用户注册成功"))
        else:
            return jsonify(common.falseReturn('', '用户注册失败'))


    # 用户登录接口已由flask-jwt默认定义好,默认路由是"/auth",可以在配置文件中配置:
    # JWT_AUTH_URL_RULE = '/login'
    # 修改登录接口路由为'/login'
    # 需要注意的是,登录接口的传值要使用 application/json 形式


    @app.route('/user')
    @jwt_required()
    def get():
        """
        获取用户信息
        :return: json
        """
        user = Users.get(Users, current_identity.id)
        returnUser = {
            'id': user.id,
            'username': user.username,
            'email': user.email,
            'login_time': user.login_time
        }
        return jsonify(common.trueReturn(returnUser, "请求成功"))

与PyJWT不同的是,flask-jwt使用装饰器的形式为需要token验证的路由做保护,从上面代码可以看到,在获取用户信息的接口中,使用@jwt_required()指定这个路由是受保护的,需要使用正确的token才能获取想要信息。其中current_identity返回当前提供token解析后的payload信息。

另外,需要注意的是,登录接口的传值要使用 application/json 形式

3、认证模块

模块入口(空)


授权认证处理:

from flask_jwt import JWT, jwt_required
from app.users.model import Users
from .. import config
from .. import common
import time

class Auth():
    def error_handler(self, e):
        print(e)
        return "Something bad happened", 400


    def authenticate(self, username, password):
        userInfo = Users.query.filter_by(username=username).first()
        if (userInfo is None):
            self.error_handler('找不到用户')
        else:
            if (Users.check_password(Users, userInfo.password, password)):
                login_time = int(time.time())
                userInfo.login_time = login_time
                Users.update(Users)
                return userInfo
            else:
                self.error_handler('密码不正确')


    def identity(self, payload):
        id = payload['identity']
        return Users.get(Users, id)

Flask-jwt对JWT生成和解析的过程进行了封装,所以使用时只需要调用它的接口,从上面的代码可以看到,实现代码非常简洁。

(1)错误处理

认证类的error_handler方法提供了异常处理,flask-jwt内部也封装了对异常和各类错误的处理,可以从后面运行的结果中看到。

(2)认证方法

认证类的authenticate方法提供认证服务,正确返回用户信息用作生成token的用户依据(payload)

(3)鉴权

认证类的identity用于鉴权并返回结果

 

四、运行结果

1、注册成功

2、注册失败

3、登录成功

4、登录失败

5、成功获取用户信息

6、获取用户信息失败

 

Flask-jwt的使用也比较简单,其本身就作了较多的封装,实现起来需要我们做的事情并不多,但也正因为如此,个人觉得失去了灵活性,与PyJWT相比,我个人倾向于PyJWT。

本文完。

 

那时那我

随遇,随缘,随安,随喜!

6 thoughts to “Flask + flask-jwt 实现基于Json Web Token的用户认证授权”

发表评论

电子邮件地址不会被公开。 必填项已用*标注

This site uses Akismet to reduce spam. Learn how your comment data is processed.