将JWT身份验证（Flask-JWT-Extended）集成到Python的Flask Web API中

首先之前，我在Python中创建的Flask Web API中添加了JWT（JSON Web Token）身份验证。但是，当我将Python版本从3.8升级到3.11时，出现了错误。因此，我尝试切换到Flask-JWT-Extended替代flask-jwt。

将JWT身份验证集成到Python的Flask Web API中。

无法从’collections’导入名为’Mapping’的名字。我将Python版本从3.8系更新到3.11系时发生了这个错误。

...
from collections import Mapping
ImportError: cannot import name 'Mapping' from 'collections' 
...

从Python 3.3开始，collections.Mapping被标记为不推荐使用，并且在Python 3.10中已从collections中移除。另外，由于Flask-JWT长时间未进行维护，它使用了collections.Mapping，这是导致问题的原因。

所以，我匆忙地在Python 3.11版本中寻找了类似的解决方案，并最终选择了Flask-JWT-Extended。

Flask-JWT-Extended 是什么？Flask-JWT-Extended 用汉语进行简述：

Flask-JWT-Extended不仅为Flask添加了对使用JSON Web Tokens (JWT)进行路由保护的支持，还内置了许多有用的（可选）功能，使得处理JSON Web Tokens更加便捷。

Flask-JWT-Extended 不仅为 Flask 添加了支持以使用 JWT 来保护路由，还添加了许多（以及可选的）功能，以便更轻松地操作 JWT。

环境

• ローカル環境

Windows 11 Pro 22H2
Python 3.11.1
PowerShell 7.3.1
Visual Studio Code 1.74.3
Git for Windows 2.39.1.windows.1
MongoDB 6.0.3 / Mongosh 1.6.0

准备用户信息为了确定用户能否使用Web API的功能，我们在MongoDB中添加了一个名为users的集合，并注册了用户信息，通过用户名称和密码来判断。

> mongosh localhost:27017/admin -u admin -p
> use holoduledb
switched to db holoduledb
> show collections
holodules
> db.createCollection("users");
{ "ok" : 1 }
> db.users.insertOne( {"id":"1", "username":"user01", "password":"dummy", "firstname": "taro", "lastname": "tokyo"} );
{
  acknowledged: true,
  insertedId: ObjectId("63d1dfafcec12c32af27ec11")
}
> db.users.find();
[
  {
    _id: ObjectId("63d1dfafcec12c32af27ec11"),
    id: '1',
    username: 'user01',
    password: 'dummy',
    firstname: 'taro',
    lastname: 'tokyo'
  }
]

添加一个用于在 Flask 中使用 JWT 身份验证的包。请先添加 Flask-JWT-Extended 包。

> poetry add flask-jwt-extended

将JWT认证处理嵌入我們將JWT身份驗證功能整合到現有的app.py網絡API中。

添加已创建的User类和Flask-JWT包的导入

from flask_jwt_extended import jwt_required, create_access_token, JWTManager, get_jwt_identity
from models.user import User

写出JWT的设置
密钥 JWT_SECRET_KEY 是从配置文件中设置的。

app.config['JWT_SECRET_KEY'] = settings.jwt_secret_key      # JWTに署名する際の秘密鍵
app.config['JWT_ALGORITHM'] = 'HS256'                       # 暗号化署名のアルゴリズム
app.config['JWT_LEEWAY'] = 0                                # 有効期限に対する余裕時間
app.config['JWT_EXPIRATION_DELTA'] = timedelta(seconds=300) # トークンの有効期間
app.config['JWT_NOT_BEFORE_DELTA'] = timedelta(seconds=0)   # トークンの使用を開始する相対時間

添加JWT验证错误处理程序

@log(logger)
def jwt_unauthorized_loader_handler(reason):
    logger.error(f"{reason}")
    return make_response(jsonify({'error': 'Unauthorized'}), 401)

将应用程序与 flask_jwt_extended 进行关联

jwt = JWTManager(app)
jwt.unauthorized_loader(jwt_unauthorized_loader_handler)

添加一个函数，通过用户名和密码进行身份验证，并返回访问令牌。

@log(logger)
@app.route('/login', methods=['POST'])
def login():
    if not request.is_json:
        abort(400)

    request_body = request.get_json()
    if request_body is None:
        abort(400)

    whitelist = {'username', 'password'}
    if not request_body.keys() <= whitelist:
        abort(400)

    user = User.from_doc(db.users.find_one({"username": request_body['username']}))
    authenticated = True if user is not None and user.password == request_body['password'] else False
    auth_user = user if authenticated else None

    if auth_user is None:
        abort(401)

    access_token = create_access_token(identity=auth_user.username)
    response_body = {'access_token': access_token}
    return make_response(jsonify(response_body), 200)

在需要JWT认证的Web API方法上指定装饰器。

@log(logger)
@app.route('/holodules/<string:date>', methods=['GET'])
@jwt_required()
def holodules(date):
    logger.info(f"username: {get_jwt_identity()}")
    logger.info(f"date: {date}")

    if len(date) != 8:
        abort(500)
    ...

JWT 身份验证处理的操作确认

启动 Web API

> poetry run python app.py
 * Serving Flask app 'app'
 * Debug mode: off
[INFO    ]_log - WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on http://127.0.0.1:5000
[INFO    ]_log - Press CTRL+C to quit

使用邮递员(Postman)，以获取访问令牌(access token)。请确认能够通过提供用户名和密码来向/login发送请求并获取访问令牌。

Action : POST
URL : http://127.0.0.1:5000/login
Headers : Content-Type: application/json
Body(raw) : {"username": "user01", "password": "password01"}

{
    "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
}

使用邮差工具调用 Web API 的方法
通过指定获取的访问令牌，调用Web API方法并确保可以获取到响应。

Action : POST
URL : http://127.0.0.1:5000/holodules/20230126
Headers : Content-Type: application/json, Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

[
    {
        "key": "HL0501_20230126_210000",
        "code": "HL0501",
        "video_id": "yMNNSGV6eT0",
        "datetime": "20230126 210000",
        "name": "獅白ぼたん",
        "title": "配信予定地【獅白ぼたん/ホロライブ】",
        "url": "https://www.youtube.com/watch?v=yMNNSGV6eT0",
        "description": "お外中なので帰宅したらあらためてつくる～！✨メンバーシップ参加はこちらから✨https://www.youtube.com/channel/UCUKD-uaobj9jiqB-VXt71mA/join特"
    },
    {
        "key": "HL0004_20230126_190100",
        "code": "HL0004",
        "video_id": "JKm_iWxBghE",
        "datetime": "20230126 190100",
        "name": "星街すいせい",
        "title": "新髪形お披露目＋ちょっと告知！【ホロライブ / 星街すいせい】",
        "url": "https://www.youtube.com/watch?v=JKm_iWxBghE",
        "description": "?2023/1/25 星街すいせい 2ndアルバム『Specter』発売！?2023/1/28 Hoshimachi Suisei 2nd Solo Live Shout in Crisis 開催！◆"
    },
    ...
]

app.py 中已集成 Flask-JWT-Extended。

import json
from flask import Flask, jsonify, request, abort, make_response, current_app
from flask import jsonify, request, Flask
from flask_jwt_extended import jwt_required, create_access_token, JWTManager, get_jwt_identity
from flask_cors import CORS
from pymongo import MongoClient
from os.path import join, dirname
from urllib.parse import quote_plus
from datetime import timedelta
from models.holodule import Holodule
from models.user import User
from settings import Settings
from logger import log, get_logger

# ロギングの設定
json_path = join(dirname(__file__), "config/logger.json")
log_dir = join(dirname(__file__), "log")
logger = get_logger(log_dir, json_path, False)

# Settings インスタンス
settings = Settings(join(dirname(__file__), '.env'))

# MongoDB 接続情報
mongodb_user = quote_plus(settings.mongodb_user)
mongodb_password = quote_plus(settings.mongodb_password)
mongodb_host = "mongodb://%s/" % (settings.mongodb_host)

# MongoDB 接続認証
client = MongoClient(mongodb_host)
db = client.holoduledb
db.authenticate(name=mongodb_user,password=mongodb_password)

# Flask
app = Flask(__name__)
app.url_map.strict_slashes = False

# CORS
CORS(app)

# JSONのソートを抑止
app.config['JSON_SORT_KEYS'] = False

# Flask JWT
app.config['JWT_SECRET_KEY'] = settings.jwt_secret_key      # JWTに署名する際の秘密鍵
app.config['JWT_ALGORITHM'] = 'HS256'                       # 暗号化署名のアルゴリズム
app.config['JWT_LEEWAY'] = 0                                # 有効期限に対する余裕時間
app.config['JWT_EXPIRATION_DELTA'] = timedelta(seconds=300) # トークンの有効期間
app.config['JWT_NOT_BEFORE_DELTA'] = timedelta(seconds=0)   # トークンの使用を開始する相対時間

# JWT の認証エラーハンドラ
@log(logger)
def jwt_unauthorized_loader_handler(reason):
    logger.error(f"{reason}")
    return make_response(jsonify({'error': 'Unauthorized'}), 401)

# JWT
jwt = JWTManager(app)
jwt.unauthorized_loader(jwt_unauthorized_loader_handler)

# レスポンスにCORS許可のヘッダーを付与
@app.after_request
def after_request(response):
    # response.headers.add('Access-Control-Allow-Origin', '*')
    response.headers.add('Access-Control-Allow-Headers', 'Content-Type,Authorization')
    response.headers.add('Access-Control-Allow-Methods', 'GET,PUT,POST,DELETE,OPTIONS')
    return response

# ログインしてトークンを返却
@log(logger)
@app.route('/login', methods=['POST'])
def login():
    if not request.is_json:
        abort(400)

    request_body = request.get_json()
    if request_body is None:
        abort(400)

    whitelist = {'username', 'password'}
    if not request_body.keys() <= whitelist:
        abort(400)

    u = request_body['username']
    p = request_body['password']

    user = User.from_doc(db.users.find_one({"username": request_body['username']}))
    authenticated = True if user is not None and user.password == request_body['password'] else False
    auth_user = user if authenticated else None

    if auth_user is None:
        abort(401)

    access_token = create_access_token(identity=auth_user.username)
    response_body = {'access_token': access_token}
    return make_response(jsonify(response_body), 200)

# ホロジュール配信予定を取得
@log(logger)
@app.route('/holodules/<string:date>', methods=['GET'])
@jwt_required()
def holodules(date):
    logger.info(f"username: {get_jwt_identity()}")
    logger.info(f"date: {date}")

    if len(date) != 8:
        abort(500)

    # MongoDB から年月日を条件にホロジュール配信予定を取得してリストに格納
    holodule_list = []
    for holodule in db.holodules.find({"datetime": {'$regex':'^'+date}}).sort("datetime", -1):
        holodule_list.append(Holodule.from_doc(holodule))

    # オブジェクトリストをJSON配列に変換
    holodules = []
    for holodule in holodule_list:
        holodules.append(holodule.to_doc())

    # UTF-8コードの application/json として返却
    return make_response(jsonify(holodules), 200)

# エラーハンドラ：400
@log(logger)
@app.errorhandler(400)
def bad_request(error):
    logger.error(f"{error}")
    return make_response(jsonify({'error': 'Bad request'}), 400)

# エラーハンドラ：401
@log(logger)
@app.errorhandler(401)
def Unauthorized(error):
    logger.error(f"{error}")
    return make_response(jsonify({'error': 'Unauthorized'}), 401)

# エラーハンドラ：404
@log(logger)
@app.errorhandler(404)
def not_found(error):
    logger.error(f"{error}")
    return make_response(jsonify({'error': 'Not found'}), 404)

# エラーハンドラ：500
@log(logger)
@app.errorhandler(500)
def internal_server_error(error):
    logger.error(f"{error}")
    return make_response(jsonify({'error': 'Internal Server Error'}), 500)

if __name__ == "__main__":
    app.run()

用于日志输出的 logger.py

import json
import datetime
import inspect
from os.path import join
from functools import wraps
from logging import config, getLogger, Filter

class CustomFilter(Filter):
    def filter(self, record):
        record.real_filename = getattr(record, 'real_filename', record.filename)
        record.real_funcName = getattr(record, 'real_funcName', record.funcName)
        record.real_lineno = getattr(record, 'real_lineno', record.lineno)
        return True

def get_logger(log_dir, json_path, verbose=False):
    with open(json_path, "r", encoding="utf-8") as f:
        log_config = json.load(f)
    # ログファイル名を日付とする
    log_path = join(log_dir, f"{datetime.datetime.now().strftime('%Y%m%d')}.log")
    log_config["handlers"]["rotateFileHandler"]["filename"] = log_path
    # verbose引数が設定されていればレベルをINFOからDEBUGに置換
    if verbose:
        log_config["root"]["level"] = "DEBUG"
        log_config["handlers"]["consoleHandler"]["level"] = "DEBUG"
        log_config["handlers"]["rotateFileHandler"]["level"] = "DEBUG"
    # ロギングの設定を適用してロガーを取得
    config.dictConfig(log_config)
    logger = getLogger(__name__)
    #logger.addFilter(CustomFilter())
    return logger

def log(logger):
    def _decorator(func):
        # funcのメタデータを引き継ぐ
        @wraps(func)
        def wrapper(*args, **kwargs):
            func_name = func.__name__
            extra = {
                'real_filename': inspect.getfile(func),
                'real_funcName': func_name,
                'real_lineno': inspect.currentframe().f_back.f_lineno
            }
            # funcの開始
            logger.info(f'[START] {func_name}', extra=extra)
            try:
                # funcの実行
                return func(*args, **kwargs)
            except Exception as err:
                # funcのエラーハンドリング
                logger.error(err, exc_info=True, extra=extra)
            finally:
                # funcの終了
                logger.info(f'[END] {func_name}', extra=extra)
        return wrapper
    return _decorator

从 .env 文件中获取设置信息的 settings.py

import os
from dotenv import load_dotenv

class Settings:
    def __init__(self, envpath):
        # .env ファイルを明示的に指定して環境変数として読み込む
        self.__dotenv_path = envpath
        load_dotenv(self.__dotenv_path)
        # 環境変数から設定値を取得
        self.__mongodb_user = os.environ.get("MONGODB_USER")
        self.__mongodb_password = os.environ.get("MONGODB_PASSWORD")
        self.__mongodb_host = os.environ.get("MONGODB_HOST")
        self.__jwt_secret_key = os.environ.get("JWT_SECRET_KEY")

    # mongodb の ユーザー
    @property
    def mongodb_user(self):
        return self.__mongodb_user

    # mongodb の パスワード
    @property
    def mongodb_password(self):
        return self.__mongodb_password

    # mongodb の ホスト:ポート
    @property
    def mongodb_host(self):
        return self.__mongodb_host

    # JWT の秘密鍵
    @property
    def jwt_secret_key(self):
        return self.__jwt_secret_key

在 models/user.py 中描写了用户信息的定义。

class User:
    def __init__(self, id, username, password, firstname, lastname):
        self.__id = id
        self.__username = username
        self.__password = password
        self.__firstname = firstname
        self.__lastname = lastname

    # id
    @property
    def id(self):
        return self.__id

    @id.setter
    def id(self, id):
        self.__id = id

    # username
    @property
    def username(self):
        return self.__username

    @username.setter
    def username(self, username):
        self.__username = username

    # password
    @property
    def password(self):
        return self.__password

    @password.setter
    def password(self, password):
        self.__password = password

    # firstname
    @property
    def firstname(self):
        return self.__firstname

    @firstname.setter
    def firstname(self, firstname):
        self.__firstname = firstname

    # lastname
    @property
    def lastname(self):
        return self.__lastname

    @lastname.setter
    def lastname(self, lastname):
        self.__lastname = lastname

    # ドキュメントから変換
    @classmethod
    def from_doc(cls, doc):
        if doc is None:
            return None
        user = User(doc['id'], 
                    doc['username'], 
                    doc['password'],
                    doc['firstname'],
                    doc['lastname'])
        return user

    # ドキュメントへ変換
    def to_doc(self):
        doc = { 'id': str(self.id),
                'username': str(self.username),
                'password' : str(self.password),
                'firstname' : str(self.firstname),
                'lastname' : str(self.lastname) }
        return doc

在 models/holodule.py 文件中记录了与 Holodule 相关的独特定义。

import datetime

class Holodule:
    codes = {
        "ホロライブ" : "HL0000",
        "ときのそら"  : "HL0001",
        "ロボ子さん" : "HL0002",
        "さくらみこ" : "HL0003",
        "星街すいせい" : "HL0004",
        "AZKi" : "HL0005",
        "夜空メル" : "HL0101",
        "アキ・ローゼンタール" : "HL0102",
        "赤井はあと" : "HL0103",
        "白上フブキ" : "HL0104",
        "夏色まつり" : "HL0105",
        "湊あくあ" : "HL0201",
        "紫咲シオン" : "HL0202",
        "百鬼あやめ" : "HL0203",
        "癒月ちょこ" : "HL0204",
        "大空スバル" : "HL0205",
        "大神ミオ" : "HL0G02",
        "猫又おかゆ" : "HL0G03",
        "戌神ころね" : "HL0G04",
        "兎田ぺこら" : "HL0301",
        "潤羽るしあ" : "HL0302",
        "不知火フレア" : "HL0303",
        "白銀ノエル" : "HL0304",
        "宝鐘マリン" : "HL0305",
        "天音かなた" : "HL0401",
        "桐生ココ" : "HL0402",
        "角巻わため" : "HL0403",
        "常闇トワ" : "HL0404",
        "姫森ルーナ" : "HL0405",
        "獅白ぼたん" : "HL0501",
        "雪花ラミィ" : "HL0502",
        "尾丸ポルカ" : "HL0503",
        "桃鈴ねね" : "HL0504",
        "魔乃アロエ" : "HL0505",
        "ラプラス" : "HL0601",
        "鷹嶺ルイ" : "HL0602",
        "博衣こより" : "HL0603",
        "沙花叉クロヱ" : "HL0604",
        "風真いろは" : "HL0605"        
    }

    def __init__(self, code="", video_id="", datetime=None, name="", title="", url="", description=""):
        self.__code = code
        self.__video_id = video_id
        self.__datetime = datetime
        self.__name = name
        self.__title = title
        self.__url = url
        self.__description = description

    # キー
    @property
    def key(self):
        _code = self.code;
        _code = Holodule.codes[self.name] if self.name in Holodule.codes else ""
        _dttm = self.datetime.strftime("%Y%m%d_%H%M%S") if self.datetime is not None else ""
        return _code + "_" + _dttm if ( len(_code) > 0 and len(_dttm) > 0 ) else ""

    # コード
    @property
    def code(self):
        return self.__code

    # video_id
    @property
    def video_id(self):
        return self.__video_id

    @video_id.setter
    def video_id(self, video_id):
        self.__video_id = video_id

    # 日時
    @property
    def datetime(self):
        return self.__datetime

    @datetime.setter
    def datetime(self, datetime):
        self.__datetime = datetime

    # 名前
    @property
    def name(self):
        return self.__name

    @name.setter
    def name(self, name):
        self.__name = name

    # タイトル（Youtubeから取得）
    @property
    def title(self):
        return self.__title

    @title.setter
    def title(self, title):
        self.__title = title

    # URL
    @property
    def url(self):
        return self.__url

    @url.setter
    def url(self, url):
        self.__url = url

    # 説明（Youtubeから取得）
    @property
    def description(self):
        return self.__description

    @description.setter
    def description(self, description):
        self.__description = description

    # ドキュメントから変換
    @classmethod
    def from_doc(cls, doc):
        if doc is None:
            return None
        holodule = Holodule(doc['code'] if 'code' in doc else Holodule.codes[doc['name']],
                            doc['video_id'], 
                            datetime.datetime.strptime(doc['datetime'], '%Y%m%d %H%M%S'), 
                            doc['name'], 
                            doc['title'], 
                            doc['url'], 
                            doc['description'])
        return holodule

    # ドキュメントへ変換
    def to_doc(self):
        doc = { 'key': str(self.key),
                'code': str(self.code),
                'video_id': str(self.video_id),
                'datetime' : str(self.datetime.strftime("%Y%m%d %H%M%S")),
                'name' : str(self.name),
                'title' : str(self.title),
                'url' : str(self.url),
                'description' : str(self.description) }
        return doc

最后
在使用 React 进行 Web 应用开发时，由于更新了后端 Python 的版本，导致了这次的错误，因此我觉得这正是一个合适的时机来对整体进行重新审视。

由于更新了收集数据的Python程序版本，它似乎会在一段时间内稳定运行。