跳转至

Flask 整合 JWT

约 108 个字 56 行代码 预计阅读时间 1 分钟

安装包

1
pip install flask-jwt-extended

整合

配置类
1
2
3
class FlaskPlaygroundConfig:
    # ...
    JWT_SECRET_KEY = '自己随便输入，不可外泄'
app.py
1
2
3
app = Flask(__name__)
app.config.from_object(FlaskPlaygroundConfig)
jwt = JWTManager(app)

基本使用方式

创建 token

假设读取到了用户信息。

1
access_token = create_access_token(identity=str(user.id))

注意必须传入字符串，否则解密会报错。

查验 token

1
2
3
4
5
@route_bp.route('/', methods=['PUT'])
@jwt_required() # 添加该装饰器后，该路由必须有对应 token 才能访问
def create_route():
    current_user_id = get_jwt_identity()
    # ...

用户访问

HTTP 请求中，在请求头中添加：

1
Authorization: Bearer 创建得到的access_token

即可访问受保护的内容。

与 SQLAlchemy 联动，从传入 / 出 ID 改为传入 / 出用户实体类

假设用户实体类如下：

1
2
3
4
5
6
7
8
9
class UserEntity(db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    create_time = db.Column(db.DateTime, default=datetime.now)
    create_user = db.Column(db.Integer, default=0)
    is_deleted = db.Column(db.Boolean, default=False)
    username = db.Column(db.String(20), unique=True)
    nickname = db.Column(db.String(20), nullable=False, default='Default Nickname')

配置

ap.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
app = Flask(__name__)
app.config.from_object(FlaskPlaygroundConfig)
jwt = JWTManager(app)

# Register a callback function that takes whatever object is passed in as the
# identity when creating JWTs and converts it to a JSON serializable format.
@jwt.user_identity_loader
def user_identity_lookup(user):
    return str(user.id)


# Register a callback function that loads a user from your database whenever
# a protected route is accessed. This should return any python object on a
# successful lookup, or None if the lookup failed for any reason (for example
# if the user has been deleted from the database).
@jwt.user_lookup_loader
def user_lookup_callback(_jwt_header, jwt_data):
    identity = jwt_data["sub"]
    return UserEntity.query.filter_by(id=identity, is_deleted=False).one_or_none()

创建 token

auth.py
1
2
3
4
5
6
7
8
@auth_bp.route('/login', methods=['POST'])
def login():
    # ...
    user = db.session.query(UserEntity).filter(UserEntity.username == username).filter(UserEntity.password == UserEntity.generate_password(password)).first()
    if not user:
        return 'username or password error'
    access_token = create_access_token(identity=user)
    # ...

查验 token

route.py
1
2
3
4
5
6
@route_bp.route('/', methods=['PUT'])
@jwt_required()
def create_route():
    # current_user 即为 token 对应的用户实例
    user_id = current_user.id
    # ...