33. 应用练习:第13版
版本 13 对版本 12 进行了以下修改:
- 代码的某些部分已进行重构(重构);
- 会话管理改用 [flask_session] 模块处理;
- 配置文件中使用了加密密码;
版本 13 最初是通过复制版本 12 创建的:


- 在 [1] 中,将对配置进行重构。具体而言,它将移出 [flask] 文件夹;
- 在 [2] 中,主脚本将进行重构。它也将移出 [flask] 文件夹;
- 在 [3] 中,配置将被拆分为多个文件;
- 在 [4] 中:我们将通过将代码移至其他文件来简化主脚本 [main];
- 在 [5] 中,由于用户密码现在将被加密,因此将修改身份验证控制器;
- 在 [6] 中,主控制器将整合此前位于主脚本 [main] 中的代码;
33.1. 重构应用程序配置

创建了三个新的配置文件:
- [mvc]:用于配置应用程序的 MVC 架构;
- [parameters]:用于存放应用程序的所有常量;
- [syspath]:用于配置应用程序的 Python 路径;
[syspath.py] 文件内容如下:
| def configure(config: dict) -> dict:
import os
# folder of this file
script_dir = os.path.dirname(os.path.abspath(__file__))
# root path
root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
# dependencies
absolute_dependencies = [
# project files
# BaseEntity, MyException
f"{root_dir}/classes/02/entities",
# InterfaceImpôtsDao, InterfaceImpôtsMétier, InterfaceImpôtsUi
f"{root_dir}/impots/v04/interfaces",
# AbstractImpôtsdao, ImpôtsConsole, ImpôtsMétier
f"{root_dir}/impots/v04/services",
# ImpotsDaoWithAdminDataInDatabase
f"{root_dir}/impots/v05/services",
# AdminData, ImpôtsError, TaxPayer
f"{root_dir}/impots/v04/entities",
# Constants, slices
f"{root_dir}/impots/v05/entities",
# Logger, SendAdminMail
f"{root_dir}/impots/http-servers/02/utilities",
# main script folder
script_dir,
# configs [database, layers, parameters, controllers, views]
f"{script_dir}/../configs",
# controllers
f"{script_dir}/../controllers",
# answers HTTP
f"{script_dir}/../responses",
# view models
f"{script_dir}/../models_for_views",
]
# set the syspath
from myutils import set_syspath
set_syspath(absolute_dependencies)
# we return the configuration
return {
"root_dir": root_dir,
"script_dir": script_dir
}
|
- [syspath] 脚本用于配置应用程序的 Python 路径(第 40–41 行);
- 它提供了两项对其他配置脚本有用的信息(第 45–46 行);
[mvc] 脚本用于配置 JSON/XML/HTML Web 应用程序的 MVC 架构:
| def configure(config: dict) -> dict:
# application configuration MVC
# controllers
from AfficherCalculImpotController import AfficherCalculImpotController
from AuthentifierUtilisateurController import AuthentifierUtilisateurController
from CalculerImpotController import CalculerImpotController
from CalculerImpotsController import CalculerImpotsController
from FinSessionController import FinSessionController
from GetAdminDataController import GetAdminDataController
from InitSessionController import InitSessionController
from ListerSimulationsController import ListerSimulationsController
from MainController import MainController
from SupprimerSimulationController import SupprimerSimulationController
# answers HTTP
from HtmlResponse import HtmlResponse
from JsonResponse import JsonResponse
from XmlResponse import XmlResponse
# view models
from ModelForAuthentificationView import ModelForAuthentificationView
from ModelForCalculImpotView import ModelForCalculImpotView
from ModelForErreursView import ModelForErreursView
from ModelForListeSimulationsView import ModelForListeSimulationsView
# authorized shares and their controllers
controllers = {
# initialization of a calculation session
"init-session": InitSessionController(),
# user authentication
"authentifier-utilisateur": AuthentifierUtilisateurController(),
# tax calculation in individual mode
"calculer-impot": CalculerImpotController(),
# batch mode tax calculation
"calculer-impots": CalculerImpotsController(),
# list of simulations
"lister-simulations": ListerSimulationsController(),
# deleting a simulation
"supprimer-simulation": SupprimerSimulationController(),
# end of calculation session
"fin-session": FinSessionController(),
# display tax calculation view
"afficher-calcul-impot": AfficherCalculImpotController(),
# obtaining data from tax authorities
"get-admindata": GetAdminDataController(),
# main controller
"main-controller": MainController()
}
# different response types (json, xml, html)
responses = {
"json": JsonResponse(),
"html": HtmlResponse(),
"xml": XmlResponse()
}
# HTML views and their models depend on the state rendered by the controller
views = [
{
# authentication view
"états": [
700, # /init-session - success
201, # /authentifier-user failure
],
"view_name": "views/vue-authentification.html",
"model_for_view": ModelForAuthentificationView()
},
{
# tax calculation
"états": [
200, # /authentifier-user success
300, # /calculate-tax-success
301, # /calculate-tax failure
800, # /show-tax-calculation-success
],
"view_name": "views/vue-calcul-impot.html",
"model_for_view": ModelForCalculImpotView()
},
{
# view of simulation list
"états": [
500, # /lister-simulations success
600, # /suppress-simulation success
],
"view_name": "views/vue-liste-simulations.html",
"model_for_view": ModelForListeSimulationsView()
},
]
# view of unexpected errors
view_erreurs = {
"view_name": "views/vue-erreurs.html",
"model_for_view": ModelForErreursView()
}
# redirections
redirections = [
{
"états": [
400, # /end-session success
],
# redirection to URL
"to": "/init-session/html",
}
]
# return the MVC configuration
return {
# controllers
"controllers": controllers,
# answers HTTP
"responses": responses,
# views and models
"views": views,
# list of redirections
"redirections": redirections,
# view of unexpected errors
"view_erreurs": view_erreurs
}
|
- 第 1-101 行:此代码已知;
- 第 105–116 行:我们渲染应用程序的 MVC 配置;
[parameters] 脚本收集应用程序的常量:
| def configure(config: dict) -> dict:
# application setup
# script_dir
script_dir = config['syspath']['script_dir']
# application configuration
parameters = {
# users authorized to use the application
"users": [
{
"login": "admin",
"password": "$pbkdf2-sha256$29000$mPM.h3COkTIGYOzde68VIg$7LH5Q7rN/1hW.Xa.6rcmR6h52PntvVqd5.na7EtgQNw"
}
],
# log file
"logsFilename": f"{script_dir}/../data/logs/logs.txt",
# server config SMTP
"adminMail": {
# server SMTP
"smtp-server": "localhost",
# server port SMTP
"smtp-port": "25",
# director
"from": "guest@localhost.com",
"to": "guest@localhost.com",
# mail subject
"subject": "plantage du serveur de calcul d'impôts",
# tls to True if server SMTP requires authorization, False otherwise
"tls": False
},
# thread pause time in seconds
"sleep_time": 0,
# redis server
"redis": {
"host": "127.0.0.1",
"port": 6379
},
}
# we return the application settings
return parameters
|
- 第 13 行:用户密码现在将被加密;
- 第 34–38 行:[Redis] 服务器的配置,我们稍后将回到这一部分;
有了这些新的配置文件,[config] 脚本变为如下所示:
| def configure(config: dict) -> dict:
# syspath configuration
import syspath
config['syspath'] = syspath.configure(config)
# application setup
import parameters
config['parameters'] = parameters.configure(config)
# database configuration
import database
config["database"] = database.configure(config)
# instantiation of application layers
import layers
config['layers'] = layers.configure(config)
# configuration MVC of the [web] layer
import mvc
config['mvc'] = mvc.configure(config)
# we return the configuration
return config
|
33.2. 重构主脚本 [main]

主脚本 [main.py] 仅用于启动服务器:
| # a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
sgbd = sys.argv[1].lower()
erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
print(f"syntaxe : {syntaxe}")
sys.exit()
# configure the application
import config
config = config.configure({'sgbd': sgbd})
# dependencies
from SendAdminMail import SendAdminMail
from Logger import Logger
from ImpôtsError import ImpôtsError
import redis
# send an e-mail to the administrator
def send_adminmail(config: dict, message: str):
# send an e-mail to the application administrator
config_mail = config['parameters']['adminMail']
config_mail["logger"] = config['logger']
SendAdminMail.send(config_mail, message)
# check log file
logger = None
erreur = False
message_erreur = None
try:
# logger
logger = Logger(config['parameters']['logsFilename'])
except BaseException as exception:
# log console
print(f"L'erreur suivante s'est produite : {exception}")
# we note the error
erreur = True
message_erreur = f"{exception}"
# store the logger in the config
config['logger'] = logger
# error handling
if erreur:
# mail to administrator
send_adminmail(config, message_erreur)
# end of application
sys.exit(1)
# start-up log
log = "[serveur] démarrage du serveur"
logger.write(f"{log}\n")
# check Redis server availability
redis_client = redis.Redis(host=config["parameters"]["redis"]["host"],
port=config["parameters"]["redis"]["port"])
# ping the Redis server
try:
redis_client.ping()
except BaseException as exception:
# Redis not available
log = f"[serveur] Le serveur Redis n'est pas disponible : {exception}"
# console
print(log)
# log
logger.write(f"{log}\n")
# end
sys.exit(1)
# save client [redis] in config
config['redis_client'] = redis_client
# data recovery from tax authorities
erreur = False
try:
# admindata will be read-only application data
config["admindata"] = config["layers"]["dao"].get_admindata()
# success log
logger.write("[serveur] connexion à la base de données réussie\n")
except ImpôtsError as ex:
# we note the error
erreur = True
# error log
log = f"L'erreur suivante s'est produite : {ex}"
# console
print(log)
# log file
logger.write(f"{log}\n")
# mail to administrator
send_adminmail(config, log)
# the main thread no longer needs the logger
logger.close()
# if there has been an error, we stop
if erreur:
sys.exit(2)
# import routes from web application
import routes
routes.config=config
routes.execute(__name__)
|
- 第 56–73 行:引入了一个 Redis 服务器及其客户端;
- 第 102–104 行:应用程序的路由已提取到 [routes] 脚本中;
33.2.1. [flask_session] 和 [redis] 模块
[Redis] 服务器将用于存储用户会话。我们将使用 [flask_session] 模块来管理这些会话。该模块可以在多个位置存储用户会话。Redis 是其中之一,我们将使用它。
必须在 PyCharm 终端中安装 [flask_session] 模块:
(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\packages>pip install flask-session
Collecting flask-session
…
要与 Redis 服务器通信,我们需要一个 Redis 客户端。这将由 [redis] 模块提供,我们也将安装该模块:
(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\packages>pip install redis
Collecting redis
…
33.2.2. Redis 服务器
Redis 服务器将用于存储用户会话。[flask_session] 模块的工作原理如下:
- 每个用户都有一个会话 ID,这正是发送给客户端的内容——且仅此而已。客户端仅在首次请求后接收一次会话 Cookie。该 Cookie 包含用户的会话 ID,在客户端后续请求中该 ID 不会改变。这就是为什么服务器无需发送新的会话 Cookie;
- 此前,会话内容会被发送给客户端。现在情况将不再如此。用户的会话内容将存储在 Redis 服务器上;
Laragon 自带的 Redis 服务器默认处于禁用状态。因此,您必须首先将其启用:

- 在 [3] 中,启用 [Redis] 服务器;
- 在 [4] 中,保留端口 [6379],这是 Redis 客户端默认使用的端口;
启用 Redis 后,Laragon 服务将自动重启:

可以在命令行模式下查询 Redis 服务器。打开 Laragon 终端 [6]:

- 在 [1] 中,[redis-cli] 命令以命令行模式启动 Redis 服务器的客户端;
截至 2019 年 7 月,Redis 客户端可通过 172 个命令与服务器交互 [https://redis.io/commands#list]。其中一个命令 [command count] [2] 会显示该数字 [3]。
向 [Redis] 写入数据需使用 Redis 命令 [set 属性 值] [4]。随后可通过命令 [get 属性] [5] 检索该值。
有时可能需要清空 Redis 数据库。这可通过 [flushdb] 命令 [6] 实现。随后,若查询 [title] 属性的值 [7],将返回 [nil] 引用 [8],表明未找到该属性。您也可以使用 [exists] 命令 [9-10] 来检查属性是否存在。
要退出 Redis 客户端,请输入 [quit] 命令 [11]。
您还可以使用 Web 界面来管理 Redis 服务器上的键。为此,必须确保 Laragon Apache 服务器正在运行:

此时将显示以下界面:

- 在 [1-4] 中,是存储在 Redis 服务器上的会话之一;
33.2.3. 在主脚本 [main] 中管理 Redis 服务器
[main] 脚本通过以下方式检查 Redis 服务器是否存在:
| import redis
…
# check Redis server availability
redis_client = redis.Redis(host=config["parameters"]["redis"]["host"],
port=config["parameters"]["redis"]["port"])
# ping the Redis server
try:
redis_client.ping()
except BaseException as exception:
# Redis not available
log = f"[serveur] Le serveur Redis n'est pas disponible : {exception}"
# console
print(log)
# log
logger.write(f"{log}\n")
# end
sys.exit(1)
# save client [redis] in config
config['redis_client'] = redis_client
|
- 第 4 行:类构造函数 [redis.Redis] 创建了一个 Redis 服务器客户端。其属性(地址、端口)来自 [parameters] 脚本;
- 第 8 行:[ping] 方法检查 Redis 服务器是否存在;
- 第 9–17 行:如果 ping 失败,则记录错误并停止服务器;
- 第 20 行:将 Redis 客户端引用存储在配置中;
33.2.4. 主脚本 [main] 中的路由管理
[main] 中的路由处理仅限于以下几行:
| # import routes from web application
import routes
routes.config=config
routes.execute(__name__)
|
- 第 1 行:路由已移至 [routes] 模块中;
- 第 3 行:路由需要了解执行配置;
- 第 4 行:我们通过传入执行脚本的名称(__main__)来启动 Flask 应用程序;
路由脚本如下:
| # dependencies
import os
from flask import Flask, redirect, request, session, url_for
from flask_api import status
from flask_session import Session
# flask application
app = Flask(__name__, template_folder="../flask/templates", static_folder="../flask/static")
# application configuration
config = {}
# the front controller
def front_controller() -> tuple:
# forward the request to the main controller
main_controller = config['mvc']['controllers']['main-controller']
return main_controller.execute(request, session, config)
@app.route('/', methods=['GET'])
def index() -> tuple:
# redirect to /init-session/html
return redirect(url_for("init_session", type_response="html"), status.HTTP_302_FOUND)
# init-session
@app.route('/init-session/<string:type_response>', methods=['GET'])
def init_session(type_response: str) -> tuple:
# execute the controller associated with the action
return front_controller()
# authenticate-user
@app.route('/authentifier-utilisateur', methods=['POST'])
def authentifier_utilisateur() -> tuple:
# execute the controller associated with the action
return front_controller()
# calculate-tax
@app.route('/calculer-impot', methods=['POST'])
def calculer_impot() -> tuple:
# execute the controller associated with the action
return front_controller()
# batch tax calculation
@app.route('/calculer-impots', methods=['POST'])
def calculer_impots():
# execute the controller associated with the action
return front_controller()
# lister-simulations
@app.route('/lister-simulations', methods=['GET'])
def lister_simulations() -> tuple:
# execute the controller associated with the action
return front_controller()
# delete-simulation
@app.route('/supprimer-simulation/<int:numero>', methods=['GET'])
def supprimer_simulation(numero: int) -> tuple:
# execute the controller associated with the action
return front_controller()
# end of session
@app.route('/fin-session', methods=['GET'])
def fin_session() -> tuple:
# execute the controller associated with the action
return front_controller()
# display-calculation-tax
@app.route('/afficher-calcul-impot', methods=['GET'])
def afficher_calcul_impot() -> tuple:
# execute the controller associated with the action
return front_controller()
# get-admindata
@app.route('/get-admindata', methods=['GET'])
def get_admindata() -> tuple:
# execute the controller associated with the action
return front_controller()
def execute(name: str):
# session secret key
app.secret_key = os.urandom(12).hex()
# Flask-Session
app.config.update(SESSION_TYPE='redis',
SESSION_REDIS=config['redis_client'])
Session(app)
# if you launch the Flask application via a console script
if name == '__main__':
app.config.update(ENV="development", DEBUG=True)
app.run(threaded=True)
|
- 第 9 行:实例化 Flask 应用程序;
- 第 12 行:编写脚本时应用程序的配置尚不可知。该配置仅在执行时才确定;
- 第 20–77 行:应用程序的路由定义与上一版本相同。此部分保持不变;
- 第 14–18 行:所有路由仅调用 [front_controller] 函数。我们已移除了该函数的原始代码,现在它仅调用 Web 应用程序的主控制器;
- 第 79–89 行:[execute] 是 [main] 脚本用于启动 Web 应用程序的函数;
- 第 81 行:[flask_session] 模块使用 Flask 的密钥;
- 第 82–84 行:[flask_session] 模块的配置。这涉及向 Flask 应用程序 [app] 的 [app.config] 配置中添加 [SESSION_TYPE, SESSION_REDIS] 键:
- [SESSION_TYPE]:会话类型。有多种类型。[redis] 类型表示 [flask_session] 使用 [redis] 服务器存储用户会话。因此,我们必须定义 [SESSION_REDIS] 键,该键必须是 Redis 客户端的引用;
- 第 85 行:将 [Flask-Session] 与 Flask 应用程序关联;
- 第 86–89 行:如果第 79 行中的 [name] 参数是字符串 [__main__],则启动 Flask 应用程序;
33.3. 重构主控制器
原先位于 [main] 脚本中 [front_controller] 函数内的代码已移至主控制器:

| # import dependencies
import threading
import time
from random import randint
from flask_api import status
from werkzeug.local import LocalProxy
from InterfaceController import InterfaceController
from Logger import Logger
from SendAdminMail import SendAdminMail
def send_adminmail(config: dict, message: str):
# send an e-mail to the application administrator
config_mail = config['parameters']['adminMail']
config_mail["logger"] = config['logger']
SendAdminMail.send(config_mail, message)
# main application controller
class MainController(InterfaceController):
def execute(self, request: LocalProxy, session: LocalProxy, config: dict) -> (dict, int):
# we process the request
logger = None
action = None
type_response1 = None
try:
# path elements are retrieved
params = request.path.split('/')
# action is the 1st element
action = params[1]
# no errors at the start
erreur = False
# session type must be known prior to certain actions
type_response1 = session.get('typeResponse')
if type_response1 is None and action != "init-session":
# we note the error
résultat = {"action": action, "état": 101,
"réponse": ["pas de session en cours. Commencer par action [init-session]"]}
erreur = True
# logger
logger = Logger(config['parameters']['logsFilename'])
# we store it in a config associated with the thread
thread_config = {"logger": logger}
thread_name = threading.current_thread().name
config[thread_name] = {"config": thread_config}
# log the request
logger.write(f"[MainController] requête : {request}\n")
# the thread is interrupted if requested
sleep_time = config['parameters']['sleep_time']
if sleep_time != 0:
# pause is randomized so that some threads are interrupted and others not
aléa = randint(0, 1)
if aléa == 1:
# log before break
logger.write(f"[front_controller] mis en pause du thread pendant {sleep_time} seconde(s)\n")
# break
time.sleep(sleep_time)
# some actions require authentication
user = session.get('user')
if not erreur and user is None and action not in ["init-session", "authentifier-utilisateur"]:
# we note the error
résultat = {"action": action, "état": 101,
"réponse": [f"action [{action}] demandée par utilisateur non authentifié"]}
erreur = True
# are there any mistakes?
if erreur:
# the request is invalid
status_code = status.HTTP_400_BAD_REQUEST
else:
# execute the controller associated with the action
controller = config['mvc']['controllers'][action]
résultat, status_code = controller.execute(request, session, config)
except BaseException as exception:
# other (unexpected) exceptions
résultat = {"action": action, "état": 131, "réponse": [f"{exception}"]}
status_code = status.HTTP_400_BAD_REQUEST
finally:
pass
# we log the result sent to the customer
log = f"[MainController] {résultat}\n"
logger.write(log)
# was there a fatal error?
if status_code == status.HTTP_500_INTERNAL_SERVER_ERROR:
# send an e-mail to the application administrator
send_adminmail(config, log)
# determine the desired type of response
type_response2 = session.get('typeResponse')
if type_response2 is None and type_response1 is None:
# the session type has not yet been set - it will be jSON
type_response = 'json'
elif type_response2 is not None:
# the type of response is known and in the session
type_response = type_response2
else:
# otherwise continue to use type_response1
type_response = type_response1
# build the response to be sent
response_builder = config['mvc']['responses'][type_response]
response, status_code = response_builder \
.build_http_response(request, session, config, status_code, résultat)
# close the log file if it has been opened
if logger:
logger.close()
# we send the HTTP response
return response, status_code
|
这些代码内容我们之前都讲过。
33.4. 处理加密密码
为了处理加密密码,我们将使用 [passlib] 模块,该模块可通过 PyCharm 终端进行安装:
(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\packages>pip install passlib
Collecting passlib
…
以下是一个示例脚本,用于加密作为参数传递给它的密码:

[create_hashed_password] 脚本如下(https://passlib.readthedocs.io/en/stable/):
| import sys
# encryption function
from passlib.hash import pbkdf2_sha256
# wait for the password to be encrypted
syntaxe = f"{sys.argv[0]} password"
erreur = len(sys.argv) != 2
if erreur:
print(f"syntaxe : {syntaxe}")
sys.exit()
else:
password = sys.argv[1]
# password encryption
hash = pbkdf2_sha256.hash(password)
print(f"version cryptée de [{password}] = {hash}")
# check
correct = pbkdf2_sha256.verify(password, hash)
print(correct)
|
- 第 16 行:作为参数传递的密码被加密;
- 第 20 行:我们将作为参数传递的密码 [password] 与其加密版本 [hash] 进行比较。[verify] 函数会对密码 [password] 进行加密,并将生成的加密字符串与 [hash] 进行比较。如果两个字符串相等,则返回 True;
上述脚本可让我们获取密码 [admin] 的加密版本:
C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\Scripts\python.exe C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020/impots/http-servers/08/passlib/create_hashed_password.py admin
version cryptée de [admin] = $pbkdf2-sha256$29000$fU9pTendO6c0ZoyR8r5Xqg$5ZXywIUnbMfN2hPnBaefiuqWjEbmAY.Lu06i4dwcnek
True
第 2 行,我们在 [parameters] 脚本中填入的值:
"users": [
{
"login": "admin",
"password": "$pbkdf2-sha256$29000$fU9pTendO6c0ZoyR8r5Xqg$5ZXywIUnbMfN2hPnBaefiuqWjEbmAY.Lu06i4dwcnek"
}
],
身份验证控制器 [AuthentifierUtilisateurController] 的演变如下:
| from passlib.handlers.pbkdf2 import pbkdf2_sha256
…
# check the validity of the (user, password) pair
users = config['parameters']['users']
i = 0
nbusers = len(users)
trouvé = False
while not trouvé and i < nbusers:
trouvé = user == users[i]["login"] and pbkdf2_sha256.verify(password, users[i]["password"])
i += 1
# found?
if not trouvé:
…
|
33.5. 测试
除了使用浏览器进行测试外,您还可以使用[http-servers/07]文件夹中为第12版编写的客户端。这些客户端在第13版中也应能正常工作:

- 在 [1] 中,三个客户端都应能正常运行;
- 在 [2] 中,两项测试都应能正常运行;