はじめに

以前に、Python で作成した Flask Web API へ JWT (JSON Web Token) 認証を組み込みましたが、Python のバージョンを3.8系から3.11系に更新したところ、エラーが発生したため、flask-jwt を Flask-JWT-Extended へ切り替えてみました。

旧記事：Python の Flask Web API に JWT による認証を組み込む

cannot import name 'Mapping' from 'collections'

Python のバージョンを3.8系から3.11系に更新したところ、このようなエラーが発生しました。

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

collections.Mapping が Python 3.3 以降非推奨となり、Python 3.10 より collections から削除されていること、あわせて、Flask-JWT が長くメンテナンスされておらず collections.Mapping を利用していることが原因でした。

そこで、取り急ぎ Python 3.11 系で利用できる同じようなものを探した結果、Flask-JWT-Extended に辿り着きました。

Flask-JWT-Extended とは

Flask-JWT-Extended

Flask-JWT-Extended not only adds support for using JSON Web Tokens (JWT) to Flask for protecting routes, but also many helpful (and optional) features built in to make working with JSON Web Tokens easier.

Flask-JWT-Extended は、JWT を使用してルートを保護するためのサポートを Flask に追加するだけでなく、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 認証処理の組み込み

既存の Web API の app.py に対して、JWT 認証処理を組み込みます。

作成した 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 を利用してアクセストークンを取得

/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..."
}

Postman を利用して 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 開催！◆"
    },
    ...
]

Flask-JWT-Extended を組み込んだ app.py

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

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 のバージョンもあわせて更新したため、しばらくは安定して動作してくれそうです。