将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程序版本，它似乎会在一段时间内稳定运行。
