30. 应用练习：第12版
在本章中，我们将编写一个遵循MVC架构（模型-视图-控制器）的Web应用程序。 该应用程序将以三种格式返回响应：jSON、XML、HTML。 接下来要做的内容与之前的内容相比，复杂度有了显著提升。我们将复用迄今为止学到的绝大多数概念，并详细说明通往最终应用程序的所有步骤。
30.1. MVC 架构
我们将按以下方式实现所谓的 MVC 架构模型（模型-视图-控制器）：
客户端请求的处理流程如下：
- 1 - 请求
请求的 URL 将采用 http://machine:port/action/param1/param2/… 的形式[Contrôleur principal] 将使用配置文件将请求“路由”到正确的控制器。 为此，它将使用 URL 中的 [action] 字段。URL 和 [param1/param2/…] 的其余部分由可选参数组成，这些参数将传递给操作。 此处的MVC中的C指代字符串[Contrôleur principal, Contrôleur / Action]。如果没有任何控制器能处理所请求的操作，Web服务器将返回“未找到所请求的URL”的响应。
- 2 - 处理
- 选定的操作 [2a] 可以利用 [Contrôleur principal] 传递给它的参数 parami。这些参数可能来自两个来源：
- URL 的路径 [/param1/param2/…]，
- 客户端请求正文中提交的参数；
- 在处理用户请求时，该操作可能需要 [métier] 和 [2b] 层。一旦处理完客户端的请求，该操作可能会触发各种响应。一个典型的例子是：
- 如果请求无法正确处理，则返回错误响应；
- 否则返回确认响应；
- [Contrôleur / Action] 将向主控制器返回其响应 [2c] 以及一个状态码。这些状态码将唯一地表示应用程序当前的状态。该状态码要么是成功码，要么是错误码；
- 3 - 响应
- 根据客户端请求的响应类型（jSON、 XML 或 HTML，[Contrôleur principal] 将实例化 [3a] 以生成相应的响应类型，并要求其将响应发送给客户端。 [Contrôleur principal] 将向其传递由已执行的 [Contrôleur / Action] 提供的响应及状态码；
- 如果所需的响应类型为 jSON 或 XML，则所选响应将对 [Contrôleur / Action] 提供的响应进行格式化，并通过 [3c] 发送出去。 能够处理此响应的客户端可以是 Python 控制台脚本，也可以是托管在 HTML 页面中的 JavaScript 脚本；
- 如果期望的响应类型为 HTML，则所选响应将根据给定的状态码，从 HTML 和 [Vuei] 视图中选择 [3b] 之一。 这就是 MVC 中的 V。一个状态码对应一个唯一的视图。该视图 V 将显示已执行的 [Contrôleur / Action] 的响应。 它使用 HTML、CSS 和 JavaScript 对该响应的数据进行渲染。这些数据被称为视图的模型。这就是 MVC 中的 M。此时，客户端通常是一个浏览器；
现在，让我们明确MVC Web架构与分层架构之间的联系。根据对模型的定义，这两个概念可能相关，也可能无关。以一个单层Web应用程序MVC为例：
在上图中，每个 [Contrôleur / Action] 都集成了 [métier] 和 [dao] 层的一部分。 在 [web] 层中确实存在 MVC 架构，但整个应用程序并不具备分层架构。这里只有一个层——Web 层，它包揽了所有功能。
现在，让我们考虑一种多层Web架构：
[web]层可以在不遵循MVC模型的情况下实现。此时虽然确实存在多层架构，但Web层并未实现MVC模型。
例如，在 .NET 环境中，上述 [web] 层层可通过 ASP.NET 和 MVC 实现，从而形成一种包含 [web] 层（类型为 MVC）的分层架构。 完成上述操作后，可以将该 ASP.NET MVC 层替换为经典的 ASP.NET 层（WebForms），同时保持其余部分 （业务层、DAO、驱动程序）保持不变。 此时便形成了一个分层架构，其中 [web] 层不再属于 MVC 类型。
在 MVC 中，我们提到模型 M 即视图 V（c.a.d）所展示的数据集合。这里给出了 MVC 模型 M 的另一种定义：
许多作者认为，位于 [web] 层右侧的内容构成了 MVC 的模型 M。为避免歧义，我们可以这样表述：
- 当指代 [web] 层右侧的所有内容时，称为“领域模型”；
- 当指代视图 V 所显示的数据时，称为视图模型；
下文中，凡提及“模型”，均指视图模型。
30.2. 客户端/服务器应用程序架构
Web 应用程序将采用以下架构：
- 在 [1] 中，Web 服务器将拥有两种类型的客户端：
- 在 [2] 中，一个控制台客户端将与服务器交换 jSON 和 XML；
- 在 [3] 中，一个浏览器将接收来自服务器的 HTML 并进行显示；
- Web 服务器 [1] 保留了之前版本的 [métier] 和 [dao] 层；
- Web客户端 [2] 将进行升级，以适应 Web 应用程序的新服务 URL；
- 浏览器显示的 HTML 应用程序需完全重写；
我们将分阶段开发该应用程序：
- 我们将开发服务器端版本 jSON。 我们将使用Postman客户端依次测试服务器的各项功能；此方法使我们能够构建Web服务器的框架，而无需关注应用程序的视图（=HTML）；
- 在用Postman测试完jSON服务器后，我们将使用控制台客户端进行测试；
- 随后我们将转到服务器版本 XML。我们已经看到，从 jSON 过渡到 XML 非常简单；
- 最后我们将切换到服务器版本 HTML。我们将构建 MVC 架构并定义要显示的视图。HTML 应用程序将同时使用 Postman 客户端和传统浏览器进行测试；
30.3. 服务器代码的目录结构
- 分为 [1：整个Web服务器；
- 在 [2] 中：目前我们将忽略 [static, templates, tests_views] 文件夹，该文件夹涉及服务器的 HTML 版本。 在该文件夹之外，我们将找到主脚本 [main] 及其配置文件；
- 在 [3] 中，则是 Web 服务器的控制器。这些将是类实例；
 |  |
- 在 [4] 中，服务器的响应 HTTP 将由类进行管理；
- 在 [5] 中，我们将保留先前服务器的日志文件；
当我们构建服务器版本 HTML 时，将涉及其他文件夹：
 |  |
- 在 [6] 中，包含应用程序 HTML 的静态元素；
- 在 [7] 中，HTML 应用程序的模板被分解为视图 [9] 和视图片段 [8]；
- 在 [9] 中，实现视图模型的类；
30.4. 应用程序服务 URL
为了构建 Web 服务器，我们将按照以下步骤进行：
- 基于应用程序 HTML 的视图，我们将定义 Web 应用程序需要实现的操作。此处我们将使用实际视图，但也可以仅使用纸质视图；
- 基于这些操作，我们将定义应用程序 HTML 的服务 URL；
- 我们将通过一个提供 jSON 的服务器来实现这些 URL 服务。这样可以定义 Web 服务器的框架，而无需考虑要提供的 HTML 页面。 我们将使用 Postman 测试这些 URL 服务；
- 随后将使用控制台客户端测试我们的 jSON 服务器；
- 一旦 jSON 服务器通过验证，我们将开始编写 HTML 应用程序；
第一个视图将是身份验证视图：
- 引导至该首屏的动作将命名为 [init-session] [1]；
- 点击按钮 [Valider] 将触发操作 [authentifier-utilisateur]，并提交两个参数 [2-3]；
税款计算视图：
- 在 [1] 中，导致此视图显示的操作为 [authentifier-utilisateur]；
- 在 [2] 中，点击按钮 [Valider] 会触发操作 [calculer-impot]，并传递三个参数 [2-5]；
- 点击链接 [6] 将触发操作 [lister-simulations]，且不带参数；
- 点击链接 [7] 会触发操作 [fin-session]，且不带参数；
第三个视图是经过身份验证的用户进行的模拟：
- 在 [3] 中，导致此视图的动作是 [lister-simulations]；
- 在 [2] 中，点击链接 [Supprimer] 将触发带有一个参数的操作 [supprimer-simulation]，该参数为列表中要删除的模拟编号；
- 点击链接 [3] 将触发操作 [afficher-calcul-impot]（无参数），该操作将重新显示税款计算视图；
- 点击链接 [4] 将触发无参数的操作 [fin-session]；
基于这些初步信息，我们可以定义服务器上的不同 URL 服务：
操作 | 角色 | 执行上下文 |
/init-session | 用于设定所需响应的类型（json、xml、html） | 请求 GET 可随时发出 |
/authentifier-utilisateur | 授权或拒绝用户登录 | 请求 POST。 该请求必须包含两个POST参数 [user, password] 仅当会话类型（json、xml、html）已知时才可发出 |
/计算-税款 | 进行税款计算模拟 | 请求 POST。 该请求必须包含三个POST参数：[marié, enfants, salaire] 仅当会话类型（json、xml、html）已知且用户已通过身份验证时才可发出 |
/lister-simulations | 请求查看自会话开始以来执行的模拟列表 | 请求 GET。 仅当会话类型（json、xml、html）已知且用户已通过身份验证时，方可发出 |
/删除-模拟/编号 | 从模拟列表中删除一个模拟 | 请求 GET。 仅当会话类型（json、xml、html）已知且用户已通过身份验证时，方可发出 |
/显示-税款计算 | 显示税款计算页面 HTML | 请求 GET。 仅当会话类型（json、xml、html）已知且用户已通过身份验证时才可发出 |
/结束会话 | 结束模拟会话。 | 技术上，旧的 Web 会话将被删除，并创建一个新的会话 仅当会话类型（json、xml、html）已知且用户已通过身份验证时才可发出 |
这些不同的 URL 服务将同时用于 HTML 服务器以及 jSON 或 XML 服务器。 其中有两个 URL 仅用于后两个服务器：这些是我们在此沿用的上一版 Web 客户端/服务器的 URL：
操作 | 角色 | 执行上下文 |
/get-admindata | 返回用于计算税款的税务数据 | 请求 GET。 仅在会话类型为 json 或 xml 时使用。用户必须经过身份验证 |
/calculer-impots | 根据在 jSON 中提交的纳税人列表计算税款 | 请求 GET。 仅在会话类型为 json 或 xml 时使用。用户必须经过身份验证 |
与这些操作关联的所有控制器将采用相同的方式：
- 它们将验证其参数。这些参数位于对象中：
- 对于以 [/action/param1/param2/…] 形式存在于 URL 中的参数，可在 [request.path] 对象中找到；
- 对于在请求正文中通过 [x-www-form-urlencoded] 传输的参数，则位于对象 [request.form] 中；
- 对于在请求正文中以 jSON 形式传输的参数，则将其放入 [request.data] 对象中；
- 控制器类似于一个函数或方法，用于验证其参数的有效性。但对于控制器而言，情况要稍显复杂：
- 预期的参数可能缺失；
- 控制器获取的参数是字符串。如果预期参数是数字，则控制器必须验证该参数的字符串是否确实为数字；
- 在验证完预期参数存在且语法正确后，还需检查它们在当前执行上下文中是否有效。该上下文存在于会话中。 身份验证便是执行上下文的一个典型示例。某些操作必须在客户端通过身份验证后才能处理。通常，会话中的某个键会标记身份验证是否已通过；
- 完成上述验证后，次级控制器方可开始工作。参数验证工作至关重要。我们绝不能接受客户端在应用程序生命周期的任何时刻向我们发送任意数据。必须对应用程序的生命周期进行全面控制；
- 完成工作后，次级控制器将包含键值 [action, état, réponse] 的字典返回给调用它的主控制器：
- [action] 表示刚刚执行的操作；
- [état] 是一个三位数，表示该操作处理的结果：
- [x00] 表示处理成功；
- [x01] 表示处理失败；
- [réponse] 是结果字典，格式为 {‘响应’:对象}。该对象的结构将根据处理的操作而有所不同；
接下来，我们将逐一介绍各个控制器，或者说这些控制器所处理的不同操作，正是这些操作构成了Web应用程序运行的核心节奏。
30.5. 服务器配置
[config_database] 数据库的配置以及 [config_layers] 服务器层的配置与先前版本完全一致。[config] 文件中出现了新的信息：
def configure(config: dict) -> dict:
import os
# 步骤 1 ------
# 该文件的目录
script_dir = os.path.dirname(os.path.abspath(__file__))
# 根路径
root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
# 依赖项
absolute_dependencies = [
# 项目文件夹
# 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",
# 常量、区间
f"{root_dir}/impots/v05/entities",
# 记录器，SendAdminMail
f"{root_dir}/impots/http-servers/02/utilities",
# 脚本 [config_database, config_layers]
script_dir,
# 控制器
f"{script_dir}/../controllers",
# 答案 HTTP
f"{script_dir}/../responses",
# 视图模板
f"{script_dir}/../models_for_views",
]
# 设置 syspath
from myutils import set_syspath
set_syspath(absolute_dependencies)
# Web 服务器的依赖项
# 控制器
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
# 响应HTTP
from HtmlResponse import HtmlResponse
from JsonResponse import JsonResponse
from XmlResponse import XmlResponse
# 视图模板
from ModelForAuthentificationView import ModelForAuthentificationView
from ModelForCalculImpotView import ModelForCalculImpotView
from ModelForErreursView import ModelForErreursView
from ModelForListeSimulationsView import ModelForListeSimulationsView
# 步骤 2------
# 应用程序配置
config.update({
# 有权使用该应用程序的用户
"users": [
{
"login": "admin",
"password": "admin"
}
],
# 日志文件
"logsFilename": f"{script_dir}/../data/logs/logs.txt",
# 服务器配置 SMTP
"adminMail": {
# 服务器 SMTP
"smtp-server": "localhost",
# SMTP 服务器的端口
"smtp-port": "25",
# 管理员
"from": "guest@localhost.com",
"to": "guest@localhost.com",
# 邮件主题
"subject": "plantage du serveur de calcul d'impôts",
# 如果服务器 SMTP 需要授权，则 TLS 设为 True，否则设为 False
"tls": False
},
# 线程暂停时长（以秒为单位）
"sleep_time": 0,
# 允许的操作及其控制器
"controllers": {
# 计算会话的初始化
"init-session": InitSessionController(),
# 用户身份验证
"authentifier-utilisateur": AuthentifierUtilisateurController(),
# 个人模式下的税款计算
"calculer-impot": CalculerImpotController(),
# 批量模式下的税款计算
"calculer-impots": CalculerImpotsController(),
# 模拟列表
"lister-simulations": ListerSimulationsController(),
# 删除模拟
"supprimer-simulation": SupprimerSimulationController(),
# 结束计算会话
"fin-session": FinSessionController(),
# 显示税款计算视图
"afficher-calcul-impot": AfficherCalculImpotController(),
# 获取税务机关数据
"get-admindata": GetAdminDataController(),
# 主控制器
"main-controller": MainController()
},
# 各种响应类型（json、xml、html）
"responses": {
"json": JsonResponse(),
"html": HtmlResponse(),
"xml": XmlResponse()
},
# 视图 HTML 及其模板取决于控制器返回的状态
"views": [
{
# 身份验证视图
"états": [
# /init-session 成功
700,
# /用户认证失败
201
],
"view_name": "views/vue-authentification.html",
"model_for_view": ModelForAuthentificationView()
},
{
# 税款计算视图
"états": [
# /用户认证成功
200,
# /计算税款 成功
300,
# /计算-税款 失败
301,
# /显示-税款计算
800
],
"view_name": "views/vue-calcul-impot.html",
"model_for_view": ModelForCalculImpotView()
},
{
# 模拟列表视图
"états": [
# /列出模拟
500,
# /删除-模拟
600
],
"view_name": "views/vue-liste-simulations.html",
"model_for_view": ModelForListeSimulationsView()
}
],
# 意外错误视图
"view-erreurs": {
"view_name": "views/vue-erreurs.html",
"model_for_view": ModelForErreursView()
},
# 重定向
"redirections": [
{
"états": [
400, # /会话结束成功
],
# 重定向至
"to": "/init-session/html",
}
],
}
)
# 步骤 3------
# 数据库配置
import config_database
config["database"] = config_database.configure(config)
# 步骤 4 ------
# 应用程序层的实例化
import config_layers
config['layers'] = config_layers.configure(config)
# 生成配置
return config
- 直到第 41 行，内容与以往相同；
- 第43-66行：到达第43行时，定义了服务器的Python路径。随后即可导入项目的依赖项：
- 第 45-55 行：控制器列表；
- 第 57-60 行：响应列表 HTTP；
- 第 62-66 行：视图模板列表；
- 第 68-189 行：通过一系列常量配置应用程序；
- 第 71-98 行：这些代码行我们在之前的版本中已经见过；
- 第 101-122 行：控制器字典：
- 键是操作的名称；
- 值为负责处理该操作的控制器实例。每个控制器仅实例化一个实例（单例）。同一实例将由服务器的不同线程执行。因此，需注意每个控制器可能需要修改的共享数据；
- 第125-129行：三个可能响应（HTTP）的字典：
- 键是客户端期望的响应类型（jSON、xml、html）；
- 参数值为响应实例 HTTP。每个响应生成器仅实例化一个实例（单例）。同一生成器将由服务器的不同线程执行。因此，需注意每个生成器可能需要修改的共享数据；
- 第 132-186 行：HTML 视图的配置。目前暂不考虑这些行；
- 第 191-202 行：我们在之前的版本中已经遇到过这些行；
30.6. 客户端请求在服务器内的处理流程
我们将追踪一个客户端请求从到达服务器到返回响应 HTTP 的整个流程。该流程遵循服务器 MVC 的路径。
30.6.1. 脚本 [main]
脚本 [main] 在许多方面与前几个版本的脚本相同。尽管如此，我们仍将其完整呈现，以便大家建立良好的基础：
# 等待 mysql 或 pgres 参数
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()
# 配置应用程序
import config
config = config.configure({'sgbd': sgbd})
# 依赖项
from flask import request, Flask, session, url_for, redirect
from flask_api import status
from SendAdminMail import SendAdminMail
from myutils import json_response
from Logger import Logger
import threading
import time
from random import randint
from ImpôtsError import ImpôtsError
import os
# 向管理员发送邮件
def send_adminmail(config: dict, message: str):
# 向应用程序管理员发送邮件
config_mail = config["adminMail"]
config_mail["logger"] = config['logger']
SendAdminMail.send(config_mail, message)
# 检查日志文件
logger = None
erreur = False
message_erreur = None
try:
# 日志记录器
logger = Logger(config["logsFilename"])
except BaseException as exception:
# 控制台日志
print(f"L'erreur suivante s'est produite : {exception}")
# 记录错误
erreur = True
message_erreur = f"{exception}"
# 在配置中保存日志器
config['logger'] = logger
# 错误处理
if erreur:
# 向管理员发送邮件
send_adminmail(config, message_erreur)
# 应用程序结束
sys.exit(1)
# 启动日志
log = "[serveur] démarrage du serveur"
logger.write(f"{log}\n")
print(log)
# 税务管理数据检索
erreur = False
try:
# admindata 将作为应用程序范围内的只读数据
config["admindata"] = config["layers"]["dao"].get_admindata().asdict()
# 成功日志
logger.write("[serveur] connexion à la base de données réussie\n")
except ImpôtsError as ex:
# 记录错误
erreur = True
# 错误日志
log = f"L'erreur suivante s'est produite : {ex}"
# 控制台
print(log)
# 日志文件
logger.write(f"{log}\n")
# 发送邮件给管理员
send_adminmail(config, log)
# 主线程不再需要日志记录器
logger.close()
# 如果发生错误则停止
if erreur:
sys.exit(2)
# Flask 应用程序
app = Flask(__name__, template_folder="templates", static_folder="static")
# 会话密钥
app.secret_key = os.urandom(12).hex()
# 前端控制器
def front_controller() -> tuple:
# 处理请求
logger = None
…
@app.route('/', methods=['GET'])
def index() -> tuple:
# 重定向至 /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:
# 执行与该操作关联的控制器
return front_controller()
# 用户认证
@app.route('/authentifier-utilisateur', methods=['POST'])
def authentifier_utilisateur() -> tuple:
# 执行与该操作关联的控制器
return front_controller()
# 计算税款
@app.route('/calculer-impot', methods=['POST'])
def calculer_impot() -> tuple:
# 执行与该操作关联的控制器
return front_controller()
# 列出模拟
@app.route('/lister-simulations', methods=['GET'])
def lister_simulations() -> tuple:
# 执行与该操作关联的控制器
return front_controller()
# 删除模拟
@app.route('/supprimer-simulation/<int:numero>', methods=['GET'])
def supprimer_simulation(numero: int) -> tuple:
# 执行与该操作关联的控制器
return front_controller()
# 结束会话
@app.route('/fin-session', methods=['GET'])
def fin_session() -> tuple:
# 执行与该操作关联的控制器
return front_controller()
# 显示税费计算
@app.route('/afficher-calcul-impot', methods=['GET'])
def afficher_calcul_impot() -> tuple:
# 执行与该操作关联的控制器
return front_controller()
# 获取管理员数据
@app.route('/get-admindata/<int:numero>', methods=['GET'])
def get_admindata() -> tuple:
# 执行与该操作关联的控制器
return front_controller()
# 仅主程序
if __name__ == '__main__':
# 启动服务器
app.config.update(ENV="development", DEBUG=True)
app.run(threaded=True)
- 第 1-92 行：这些行我们之前已经遇到并解释过；
- 第 92 行：服务器将管理一个会话。因此我们需要一个密钥。对于每个用户，我们将在会话中存储两项信息：
- 用户是否已成功认证；
- 每次用户进行税款计算时，计算结果将被放入一个名为“用户模拟列表”的列表中。该列表将存储在会话中；
- 第100-151行：服务器服务列表。相关函数起到过滤作用：所有未出现在该列表中的URL请求都将被Flask服务器以[404 NOT FOUND]错误拒绝。 通过此过滤后，请求将系统地转发至由第94-98行函数[front_controller]实现的“前端控制器”，我们稍后将对此进行介绍；
- 第100-103行：[/]路由管理。 Web 应用程序的入口点将是第 107 行中的 URL。因此，在第 103 行，我们将客户端重定向至该 URL：
- 函数 [url_for] 在第 18 行被调用。该函数在此处有两个参数：
- 第一个参数是路由函数的名称，此处即第107行的函数。可以看到该函数期待一个名为[type_response]的参数，该参数表示客户端期望的响应类型（json、xml、html）；
- 第二个参数引用了第107行参数的名称 [type_response] 并为其赋值。如果还有其他参数，则需对每个参数重复此操作；
- 该操作将URL与由给定两个参数指定的函数关联起来。在此处，这将生成第106行的URL，其中参数被其值[/init-session/html]替换；
- 函数 [redirect] 已在第 18 行被导入。其作用是向客户端发送一个 HTTP 重定向头部：
- 第一个参数是客户端应被重定向到的 URL；
- 第二个参数是发给客户端的响应状态码。状态码 [status.HTTP_302_FOUND] 表示重定向；
第94-98行中的函数[front_controller] 负责对客户端请求进行初步处理：
# 前端控制器
def front_controller() -> tuple:
# 处理请求
logger = None
try:
# 日志记录
logger = Logger(config["logsFilename"])
# 将其存储在与线程关联的配置中
thread_config = {"logger": logger}
thread_name = threading.current_thread().name
config[thread_name] = {"config": thread_config}
# 记录请求
logger.write(f"[ front_controller] requête : {request}\n")
# 若收到中断请求，则中断该线程
sleep_time = config["sleep_time"]
if sleep_time != 0:
# 暂停是随机的，以便某些线程被中断而其他线程不被中断
aléa = randint(0, 1)
if aléa == 1:
# 暂停前的日志
logger.write(f"[ front_controller] mis en pause du thread pendant {sleep_time} seconde(s)\n")
# 暂停
time.sleep(sleep_time)
# 将请求转发至主控制器
main_controller = config['controllers']["main-controller"]
résultat, status_code = main_controller.execute(request, session, config)
# 记录发送给客户端的结果
log = f"[front_controller] {résultat}\n"
logger.write(log)
# 是否发生致命错误？
if status_code == status.HTTP_500_INTERNAL_SERVER_ERROR:
# 向应用程序管理员发送邮件
send_adminmail(config, log)
# 确定所需的响应类型
if session.get('typeResponse') is None:
# 会话类型尚未确定——将采用 jSON
type_response = 'json'
else:
type_response = session['typeResponse']
# 构建待发送的响应
response_builder = config["responses"][type_response]
response, status_code = response_builder \
.build_http_response(request, session, config, status_code, résultat)
# 正在发送响应
return response, status_code
except BaseException as erreur:
# 这是个意外错误——如果可能的话，记录该错误
if logger:
logger.write(f"[ front_controller] {erreur}")
# 准备向客户端发送响应
résultat = {"réponse": {"erreurs": [f"{erreur}"]}}
# 发送响应至 jSON
return json_response(résultat, status.HTTP_500_INTERNAL_SERVER_ERROR)
finally:
# 如果日志文件已被打开，则将其关闭
if logger:
logger.close()
- 第1-57行：我们熟悉这段代码。例如，这是上一版本脚本[main]中名为[main]的函数代码。唯一需要注意的是第25-26行使用的控制器：
- 第25行：从配置中获取与名称[main-controller]关联的控制器实例。具体涉及以下几行：
# Web 服务器的依赖项
# 控制器
…
from MainController import MainController
# 允许的操作及其控制器
"controllers": {
…,
# 主控制器
"main-controller": MainController()
},
- （续）
- 如上文第10行所示，请注意这里获取的是一个类实例；
- 第26行：请求控制器[MainController]处理该请求；
- 第30-45行：将控制器[MainController]返回的响应发送给客户端。我们稍后将再次讨论这些行；
函数 [front_controller] 以及类 [MainController] 的作用是处理所有请求共有的工作：
在上图所示的流程中，当前仍处于请求处理的第1阶段。主控制器[MainController]将接续执行第1阶段。
30.6.2. 主控制器 [MainController]
主控制器 [MainController] 继续由函数 [front_controller] 开始的工作：
所有控制器均实现了以下接口：[InterfaceController] [2]：
from abc import ABC, abstractmethod
from werkzeug.local import LocalProxy
class InterfaceController(ABC):
@abstractmethod
def execute(self, request: LocalProxy, session: LocalProxy, config: dict) -> (dict, int):
pass
- 接口 [InterfaceController] 仅定义了第 8 行中的唯一方法 [execute]。该方法接收三个参数：
- [request]：客户端请求；
- [session]：客户端会话；
- [config]：应用程序配置；
方法 [execute] 返回一个包含两个元素的元组：
- 第一个元素是结果字典，格式为 {‘action’: action, ‘état’:état, ‘réponse’:résultats}；
- 第二个是应返回给客户端的状态码 HTTP；
主控制器 [MainController] [1] 通过以下方式实现了 [InterfaceController] 接口：
# 导入依赖项
from flask_api import status
from werkzeug.local import LocalProxy
# Web 应用程序控制器
from InterfaceController import InterfaceController
class MainController(InterfaceController):
def execute(self, request: LocalProxy, session: LocalProxy, config: dict) -> (dict, int):
# 获取路径中的元素
params = request.path.split('/')
action = params[1]
# 错误
erreur = False
# 在执行某些操作前必须已知会话类型
type_response = session.get('typeResponse')
if type_response is None and action != "init-session":
# 记录错误
résultat = {"action": action, "état": 101,
"réponse": ["pas de session en cours. Commencer par action [init-session]"]}
erreur = True
# 某些操作需要经过身份验证
user = session.get('user')
if not erreur and user is None and action not in ["init-session", "authentifier-utilisateur"]:
# 记录错误
résultat = {"action": action, "état": 101,
"réponse": [f"action [{action}] demandée par utilisateur non authentifié"]}
erreur = True
# 是否有错误？
if erreur:
# 返回一条错误消息
return résultat, status.HTTP_400_BAD_REQUEST
else:
# 执行与该操作关联的控制器
controller = config["controllers"][action]
résultat, status_code = controller.execute(request, session, config)
return résultat, status_code
控制器 [MainController] 首先对请求的有效性进行初步验证。
- 第 11-13 行：控制器首先获取客户端请求的操作。 需要说明的是，服务端 URL 采用 [/action/param1/param2/…] 格式，而该 URL 位于 [request.path] 中；
- 第17-23行：操作[init-session]用于初始化客户端所需的响应类型（json、xml、html）。该信息被存储在会话中，并关联键[typeRéponse]。 因此，如果操作不是 [init-session]，则会话中必须包含键 [typeRéponse]，否则请求将报错；
- 第21-22行：每个控制器返回结果的结构，此处为错误结果：
- [action]：是当前操作的名称。这将使我们在记录请求结果时能够获取其名称；
- [état]：是一个三位数的状态码：
- [x00] 表示成功；
- [x01] 表示失败；
- [réponse]：是请求的响应。其具体内容因请求而异；
- 第24-30行：操作[authentifier-utilisateur]用于验证用户身份。若验证成功，则会在用户会话中设置一个密钥[user=True]。 某些服务操作 URL 仅限经过身份验证的用户访问。此处即对此进行验证；
- 第 26 行：仅 [init-session] 和 [authentifier-utilisateur] 操作可由尚未经过身份验证的用户执行；
- 第28-29行：发生错误时应返回的结果；
- 第32-34行：若发生上述两种错误中的任何一种，则向客户端发送错误响应，状态码为 HTTP 400 BAD REQUEST；
- 第35-39行：若未发生错误，则将控制权移交给负责处理当前操作的控制器。该控制器的实例可在应用程序配置中找到；
类 [MainController] 承接了函数 [front_controller] 的工作：二者共同整合了请求处理中所有可提取的部分，直至最后时刻才将请求传递给特定的控制器。 [front_controller] 函数与 [MainController] 类之间的代码划分完全是主观的。 在此，我希望保留上一版本的成果：函数 [front_controller] 此前已以 [main] 的名称存在。实际上，我们可以：
- 将所有内容放入函数 [front_controller] 中，并删除类 [MainController]；
- 将所有内容移入类 [MainController]，并删除函数 [front_controller]。我更倾向于选择这一方案，因为它能有效精简主脚本 [main] 的代码；
30.7. 针对特定操作的处理
让我们回到应用程序的 MVC 架构：
我们目前仍处于上述步骤 1。如果未发生错误，将进入步骤 2。请求已转发至处理该请求所对应操作的特定控制器。假设该操作为 [/init-session]，其路由定义为：
# 初始化会话
@app.route('/init-session/<string:type_response>', methods=['GET'])
def init_session(type_response: str) -> tuple:
# 执行与该操作关联的控制器
return front_controller()
此操作与配置 [config] 中的控制器相关：
# 授权操作及其控制器
"controllers": {
# 初始化计算会话
"init-session": InitSessionController(),
…
},
因此，控制器 [InitSessionController]（第 4 行）将接管控制。其代码如下：
from flask_api import status
from werkzeug.local import LocalProxy
from InterfaceController import InterfaceController
class InitSessionController(InterfaceController):
def execute(self, request: LocalProxy, session: LocalProxy, config: dict) -> (dict, int):
# 获取路径中的元素
dummy, action, type_response = request.path.split('/')
# 初始阶段无错误
erreur = False
# 验证响应类型
if type_response not in config['responses'].keys():
erreur = True
résultat = {"action": action, "état": 701,
"réponse": [f"paramètre [type={type_response}] invalide"]}
# 若无错误
if not erreur:
# 将会话类型存入 Flask 会话
session['typeResponse'] = type_response
résultat = {"action": action, "état": 700,
"réponse": [f"session démarrée avec le type de réponse {type_response}"]}
return résultat, status.HTTP_200_OK
else:
return résultat, status.HTTP_400_BAD_REQUEST
- 第 6 行：与其他控制器一样，控制器 [InitSessionController] 实现了接口 [InterfaceController]；
- 第10行：URL属于[/init-session/type_response]类型。获取[init-session]操作及所需的响应类型；
- 第 15 行：期望的响应类型只能是响应配置中存在的类型之一：
# 各种响应类型（json、xml、html）
"responses": {
"json": JsonResponse(),
"html": HtmlResponse(),
"xml": XmlResponse()
},
- 如果不是这种情况，则准备一个 701 错误响应（第 17 行）；
- 第20-25行：当所需的响应类型有效时；
- 第22行：将期望的响应类型存入会话。因为后续请求需要使用该信息；
- 第23-24行：准备700成功响应；
- 第25行：将成功响应返回给调用方；
- 第27行：若发生错误，将错误响应返回给调用方代码；
30.8. 服务器响应 HTTP 的生成
让我们回到应用程序的 MVC 架构：
我们刚刚介绍了步骤 1 和 2。我们遇到了三种状态代码：
- 700：/init-session 成功；
- 701：/init-session 失败；
- 101：请求无效，原因可能是会话未初始化，或者用户未通过身份验证；
让我们来分析一下在上文第 3 步中，服务器响应将如何发送给客户端。这一过程发生在脚本 [main] 的函数 [front_controller] 中：
# 前端控制器
def front_controller() -> tuple:
# 处理请求
logger = None
try:
# 日志记录
logger = Logger(config["logsFilename"])
# 将其存储在与线程关联的配置中
thread_config = {"logger": logger}
thread_name = threading.current_thread().name
config[thread_name] = {"config": thread_config}
# 记录请求
logger.write(f"[ front_controller] requête : {request}\n")
# 若收到中断请求，则中断该线程
sleep_time = config["sleep_time"]
if sleep_time != 0:
# 暂停是随机的，以便某些线程被中断而其他线程不被中断
aléa = randint(0, 1)
if aléa == 1:
# 暂停前记录日志
logger.write(f"[ front_controller] mis en pause du thread pendant {sleep_time} seconde(s)\n")
# 暂停
time.sleep(sleep_time)
# 将请求转发至主控制器
main_controller = config['controllers']["main-controller"]
résultat, status_code = main_controller.execute(request, session, config)
# 记录发送给客户端的结果
log = f"[front_controller] {résultat}\n"
logger.write(log)
# 是否发生致命错误？
if status_code == status.HTTP_500_INTERNAL_SERVER_ERROR:
# 向应用程序管理员发送邮件
send_adminmail(config, log)
# 确定所需的响应类型
if session.get('typeResponse') is None:
# 会话类型尚未确定——将采用 jSON
type_response = 'json'
else:
type_response = session['typeResponse']
# 构建待发送的响应
response_builder = config["responses"][type_response]
response, status_code = response_builder \
.build_http_response(request, session, config, status_code, résultat)
# 正在发送响应
return response, status_code
except BaseException as erreur:
# 这是个意外错误——如果可能的话，记录该错误
if logger:
logger.write(f"[ front_controller] {erreur}")
# 正在准备发给客户的响应
résultat = {"réponse": {"erreurs": [f"{erreur}"]}}
# 发送响应至 jSON
return json_response(résultat, status.HTTP_500_INTERNAL_SERVER_ERROR)
finally:
# 若日志文件已打开，则关闭
if logger:
logger.close()
- 目前位于第 26 行：主控制器已返回错误响应；
- 第27-29行：无论主控制器返回何种响应（成功或失败），该响应都会被记录在日志文件中；
- 第30-33行：与之前版本一样，如果状态码HTTP为[500 INTERNAL SERVER ERROR]，则向应用程序管理员发送一封包含错误日志的邮件；
- 第34-39行：将发送响应HTTP，控制器返回的结果将被放入该响应的正文中。我们需要知道客户端希望以何种格式（json、xml、html）接收此响应。 在会话中查找所需的响应类型。如果未找到，则将该类型任意设为 jSON；
- 第40-43行：构建响应HTTP；
在配置文件中，每种响应类型（json、xml、html）都已关联到一个类实例：
# 各种响应类型（json、xml、html）
"responses": {
"json": JsonResponse(),
"html": HtmlResponse(),
"xml": XmlResponse()
},
响应类位于服务器目录树中的 [responses] 文件夹内：
每个响应类都实现了以下接口 [InterfaceResponse]：
from abc import ABC, abstractmethod
from flask.wrappers import Response
from werkzeug.local import LocalProxy
class InterfaceResponse(ABC):
@abstractmethod
def build_http_response(self, request: LocalProxy, session: LocalProxy, config: dict, status_code: int,
résultat: dict) -> (Response, int):
pass
- 第 8-11 行：接口 [InterfaceResponse] 定义了一个唯一的方法 [build_http_response]，其参数如下：
- [request, session, config]：这是操作控制器接收的参数；
- [résultat, status_code]：这是操作控制器生成的结果；
接下来我们将介绍响应 jSON。它由以下类 [JsonResponse] 生成：
import json
from flask import make_response
from flask.wrappers import Response
from werkzeug.local import LocalProxy
from InterfaceResponse import InterfaceResponse
class JsonResponse(InterfaceResponse):
def build_http_response(self, request: LocalProxy, session: LocalProxy, config: dict, status_code: int,
résultat: dict) -> (Response, int):
# 结果：结果字典
# status_code：响应状态码 HTTP
# 返回响应 HTTP
response = make_response(json.dumps(résultat, ensure_ascii=False))
response.headers['Content-Type'] = 'application/json; charset=utf-8'
return response, status_code
我们对这个代码并不陌生，此前已多次遇到。它是模块 [myutils] 中函数 [json_response] 的代码。
30.9. 初步测试
在所研究的代码中，我们发现了三个状态代码：
- 700：/init-session 成功；
- 701：/init-session 失败；
- 101：请求无效，原因可能是会话未初始化，或者用户未通过身份验证；
我们将尝试使用会话 jSON 来触发这些状态码。
- 启动Web服务器、SGBD以及邮件服务器；
- 我们启动一个 Postman 客户端；
测试 1
首先展示一个因会话未初始化而导致的无效请求：
- [1-2]：请求 [POST http://localhost:5000/authentifier-utilisateur] 是一条有效的路由：
# 用户认证
@app.route('/authentifier-utilisateur', methods=['POST'])
def authentifier_utilisateur() -> tuple:
# 执行与该操作关联的控制器
return front_controller()
但只有在先前已通过操作 [/init-session] 初始化会话的情况下，该请求才会被接受。
执行该请求，查看服务器返回的结果：
- [1-2]：我们获得了响应 jSON。当客户端尚未指定响应类型时，服务器会使用 jSON 进行响应；
- [3-5]：响应中的字典 jSON；
- [action]：已执行的操作；
- [état]：响应的状态码。[x01] 代码表示发生错误；
- [réponse]：针对每个操作而定。此处包含一条错误消息；
现在我们尝试使用错误的响应类型初始化一个会话：
- [1-2] 是一个正确的路由：
# 初始化会话
@app.route('/init-session/<string:type_response>', methods=['GET'])
def init_session(type_response: str) -> tuple:
# 执行与该操作关联的控制器
return front_controller()
因此，该请求将进入服务器MVC的请求处理流程。然而，由于请求的会话类型不正确，该请求在处理过程中应被拒绝。
响应如下：
- 在 [4] 中，包含错误代码 [x01]；
- 在 [5] 中，显示了错误说明；
现在，初始化一个 jSON 会话：
响应如下：
现在，初始化一个 XML 会话。响应 jSON 将被以下类 [XmlResponse] 生成的响应 XML 所替换：
import xmltodict
from flask import make_response
from flask.wrappers import Response
from werkzeug.local import LocalProxy
from InterfaceResponse import InterfaceResponse
from Logger import Logger
class XmlResponse(InterfaceResponse):
def build_http_response(self, request: LocalProxy, session: LocalProxy, config: dict, status_code: int,
résultat: dict) -> (Response, int):
# 结果：结果字典
# status_code：响应状态码HTTP
# 结果：需转换为字符串的字典 XML
xml_string = xmltodict.unparse({"root": résultat})
# 返回响应 HTTP
response = make_response(xml_string)
response.headers['Content-Type'] = 'application/xml; charset=utf-8'
return response, status_code
这是我们熟悉的代码，来自共享模块 [myutils] 中的函数 [xml_response]。
我们初始化一个 XML 会话：
此时服务器的输出结果如下：
我们得到的响应与 jSON 相同，但这次响应被包装成了 XML。
30.10. 操作 [authentifier-utilisateur]
操作 [authentifier-utilisateur] 用于对希望使用税款计算应用程序的用户进行身份验证。其在脚本 [main] 中的路径定义如下：
# 用户认证
@app.route('/authentifier-utilisateur', methods=['POST'])
def authentifier_utilisateur() -> tuple:
# 执行与该操作关联的控制器
return front_controller()
服务器等待两个POST参数：
- [user]：用户 ID；
- [password]：用户密码；
授权用户列表在配置文件 [config] 中定义：
# 允许使用该应用程序的用户
"users": [
{
"login": "admin",
"password": "admin"
}
],
此处是一个单元素列表。
操作 [authentifier-utilisateur] 由以下控制器 [AuthentifierUtilisateurController] 处理：
from flask_api import status
from werkzeug.local import LocalProxy
from InterfaceController import InterfaceController
from Logger import Logger
class AuthentifierUtilisateurController(InterfaceController):
def execute(self, request: LocalProxy, session: LocalProxy, config: dict) -> (dict, int):
# 获取路径中的元素
dummy, action = request.path.split('/')
# POST 的参数
post_params = request.form
# 响应状态码 HTTP
status_code = None
# 起初没有错误
erreur = False
erreurs = []
# 需要一个包含两个参数的 POST
if len(post_params) != 2:
erreur = True
status_code = status.HTTP_400_BAD_REQUEST
erreurs.append("méthode POST requise, paramètre [action] dans l'URL, paramètres postés [user, password]")
if not erreur:
# 从 POST 中获取参数
# 参数 [user]
user = post_params.get("user")
if user is None:
erreur = True
erreurs.append("paramètre [user] manquant")
# 参数 [password]
password = post_params.get("password")
if password is None:
erreur = True
erreurs.append("paramètre [password] manquant")
# 错误？
if erreur:
status_code = status.HTTP_400_BAD_REQUEST
# 错误？
if not erreur:
# 正在验证用户名和密码的有效性
users = config['users']
i = 0
nbusers = len(users)
trouvé = False
while not trouvé and i < nbusers:
trouvé = user == users[i]["login"] and password == users[i]["password"]
i += 1
# 找到？
if not trouvé:
# 记录错误
erreur = True
status_code = status.HTTP_401_UNAUTHORIZED
erreurs.append(f"Echec de l'authentification")
else:
# 在会话中记录已找到用户
session["user"] = True
# 完成
if not erreur:
# 无错误返回
résultat = {"action": action, "état": 200, "réponse": f"Authentification réussie"}
return résultat, status.HTTP_200_OK
else:
# 返回，出现错误
return {"action": action, "état": 201, "réponse": erreurs}, status_code
- 第 14 行：获取 POST 的参数；
- 第 19 行：查询中发现的错误列表；
- 第20-24行：验证是否确实提交了两个参数；
- 第27-31行：验证是否存在参数[users]；
- 第32-36行：检查是否存在参数[password]；
- 第38-39行：如果提交的参数有误，则准备响应 HTTP 400 BAD REQUEST；
- 第40-58行：验证用户名[user, password]是否属于有权使用该应用程序的用户；
- 第51-55行：如果用户（用户名、密码）无权使用该应用程序，则准备响应 HTTP 401 UNAUTHORIZED；
- 第56-58行：若用户获授权，则使用密钥[user]在会话中记录其已通过身份验证；
需要注意的是，如果用户使用 [identifiants1] 凭据通过了身份验证，但使用 [identifiants2] 凭据的身份验证失败，那么他仍然会保持使用 [identifiants1] 凭据的已验证状态。
进行 Postman 测试：
- 启动Web服务器、SGBD以及邮件服务器；
- 使用 Postman 客户端：
- 启动 jSON 会话；
- 然后进行身份验证；
以下是几种不同情况。
情况 1：POST 未发送参数
- 在 [3-5] 中，POST 没有正文；
请求结果如下：
- 在 [2] 中，我们收到了响应 HTTP 400 BAD REQUEST；
- 将 [5] 转换后，得到错误代码 [201]；
情况 2：POST 因凭证错误
- 在 [6] 中，凭证有误；
服务器返回以下响应：
- 在 [2] 中，响应为 HTTP 401 UNAUTHORIZED；
- 在 [5] 中，返回错误响应；
情况 2：POST 且凭证正确
- 变为 [6]，凭证正确；
服务器的响应如下：
- 在 [2] 中，响应为 HTTP 200 OK；
- 在 [5] 中，成功响应；
30.11. 操作 [calculer_impot]
操作 [calculer_impot] 用于计算纳税人的税款。其在脚本 [main] 中的路径定义如下：
# 计算税款
@app.route('/calculer-impot', methods=['POST'])
def calculer_impot() -> tuple:
# 执行与该操作关联的控制器
return front_controller()
服务器等待三个POST参数：
- [marié]：是/否；
- [enfants]：纳税人的子女数量；
- [salaire]：纳税人的年薪；
[CalculerImpotController] 控制器处理 [calculer_impot] 操作：
import re
from flask_api import status
from werkzeug.local import LocalProxy
from InterfaceController import InterfaceController
from TaxPayer import TaxPayer
class CalculerImpotController(InterfaceController):
def execute(self, request: LocalProxy, session: LocalProxy, config: dict) -> (dict, int):
# 获取路径中的元素
dummy, action = request.path.split('/')
# 初始阶段无错误
erreur = False
erreurs = []
# POST 的参数
post_params = request.form
# 需要一个包含三个参数的 POST
if len(post_params) != 3:
erreur = True
erreurs.append(
"méthode POST requise avec les paramètres postés [marié, enfants, salaire]")
# 正在分析已发布的参数
if not erreur:
# 参数已匹配
marié = post_params.get("marié")
if marié is None:
erreurs.append("paramètre [marié] manquant")
else:
# 该参数是否有效？
marié = marié.lower()
if marié != "oui" and marié != "non":
erreur = True
erreurs.append(f"valeur [{marié}] invalide pour le paramètre [marié (oui/non)]")
# 参数 [enfants]
enfants = post_params.get("enfants")
if enfants is None:
erreur = True
erreurs.append("paramètre [enfants] manquant")
else:
# 该参数是否有效？
enfants = enfants.strip()
match = re.match(r"\d+", enfants)
if not match:
erreur = True
erreurs.append(f"valeur [{enfants}] invalide pour le paramètre [enfants (entier>=0)]")
# 工资参数
salaire = post_params.get("salaire")
if salaire is None:
erreur = True
erreurs.append("paramètre [salaire] manquant")
else:
# 该参数是否有效？
salaire = salaire.strip()
match = re.match(r"\d+", salaire)
if not match:
erreur = True
erreurs.append(f"valeur [{salaire}] invalide pour le paramètre [salaire (entier>=0)]")
# 错误？
if erreur:
status_code = status.HTTP_400_BAD_REQUEST
résultat = {"action": action, "état": 301, "réponse": erreurs}
# 返回结果
return résultat, status_code
# 计算税款
# 检索层 [métier] 和字典 [adminData]
métier = config["layers"]["métier"]
admin_data = config["admindata"]
# 税款计算
taxpayer = TaxPayer().fromdict({'marié': marié, 'enfants': enfants, 'salaire': salaire})
métier.calculate_tax(taxpayer, admin_data)
# 模拟编号
id_simulation = session.get('id_simulation', 0)
id_simulation += 1
session['id_simulation'] = id_simulation
# 将结果以字典形式保存到会话中 TaxPayer
simulation = taxpayer.fromdict({'id': id_simulation}).asdict()
# 将结果添加到已完成的模拟列表中，并将该列表存入会话
simulations = session.get("simulations", [])
simulations.append(simulation)
session["simulations"] = simulations
# 结果
résultat = {"action": action, "état": 300, "réponse": simulation}
status_code = status.HTTP_200_OK
# 输出结果
return résultat, status_code
- 第13行：获取当前操作的名称；
- 第17行：将错误累积到列表中；
- 第19行：获取已提交的参数。这些参数以[x-www-form-urlencoded]的形式提交，因此我们在[request.form]中获取它们。 如果它们是以 jSON 的形式提交的，那么我们会将其读取为 [request.data]；
- 第21-24行：验证确实有三个参数被提交；
- 第27-36行：验证提交的参数[marié]是否存在且有效；
- 第37-48行：验证提交参数[enfants]是否存在及其有效性；
- 第49-60行：验证提交参数[salaire]是否存在及其有效性；
- 第62-66行：如果出现错误，则发送错误响应400 BAD REQUEST，并附带状态码[301]；
- 第69-71行：若未发生错误，则准备计算税款。为此，
- 第70行：从[métier]层获取引用；
- 第71行：从服务器配置中获取税务部门的数据；
- 第72-74行：计算纳税人的税款；
- 第75-77行：统计用户进行的税款计算次数；
- 第76行：从会话中获取上次计算的编号。此处将计算结果命名为[simulation]；
- 第 77 行：递增上次模拟的编号；
- 第78行：将该编号存入会话；
- 第79-84行：为追踪用户进行的计算，将用户已执行的模拟列表存入其会话中；
- 第 80 行：一个模拟将作为对象 TaxPayer 的字典，该对象的属性 [id] 的值即为该模拟的编号；
- 第82-84行：将当前模拟添加到会话中的模拟列表中；
- 第86-87行：准备一个成功响应HTTP；
- 第90行：返回结果；
进行一些测试：启动Web服务器、SGBD、邮件服务器以及Postman客户端。
情况1：在会话未初始化时进行税款计算
响应如下：
情况2：未经过身份验证时进行税款计算
首先使用 [/init-session/json] 启动会话 jSON。然后执行与之前相同的请求。此时响应如下：
情况 3：在参数缺失的情况下进行税款计算
初始化会话 jSON，完成身份验证后执行以下请求：
- 在 [5] 中，缺少参数 [marié]；
响应如下：
情况 4：使用错误参数进行税款计算
服务器的响应如下：
情况4：使用正确参数进行税款计算
服务器的响应如下：
30.12. 操作 [lister-simulations]
操作 [lister-simulations] 允许用户查看自本次会话开始以来所进行的模拟列表。其路径在脚本 [main] 中定义如下：
# 列出模拟
@app.route('/lister-simulations', methods=['GET'])
def lister_simulations() -> tuple:
# 执行与该操作关联的控制器
return front_controller()
服务器不接受任何参数。[lister-simulations] 操作由以下 [ListerSimulationsController] 控制器处理：
from flask_api import status
from werkzeug.local import LocalProxy
from InterfaceController import InterfaceController
class ListerSimulationsController(InterfaceController):
def execute(self, request: LocalProxy, session: LocalProxy, config: dict) -> (dict, int):
# 获取路径中的元素
dummy, action = request.path.split('/')
# 获取会话中的模拟列表
simulations = session.get("simulations", [])
# 返回结果
return {"action": action, "état": 500,
"réponse": simulations}, status.HTTP_200_OK
- 第 13 行：从会话中获取模拟列表；
- 第 15-16 行：返回成功响应；
进行以下 Postman 测试：
- 启动 jSON 会话；
- 进行身份验证；
- 执行两次税款计算；
- 请求模拟列表；
请求如下：
- 在 [3] 中，没有参数；
服务器的响应如下：
- 在 [4] 中，显示用户的模拟列表；
30.13. 操作 [supprimer-simulation]
操作 [supprimer-simulation] 允许用户从其模拟列表中删除某项模拟。该操作的路径在脚本 [main] 中定义如下：
# 删除模拟
@app.route('/supprimer-simulation/<int:numero>', methods=['GET'])
def supprimer_simulation(numero: int) -> tuple:
# 执行与该操作关联的控制器
return front_controller()
服务器仅接受一个参数，即要删除的模拟编号。[supprimer-simulation] 操作由以下 [SupprimerSimulationController] 控制器处理：
from flask_api import status
from werkzeug.local import LocalProxy
from InterfaceController import InterfaceController
class SupprimerSimulationController(InterfaceController):
def execute(self, request: LocalProxy, session: LocalProxy, config: dict) -> (dict, int):
# 获取路径中的元素
dummy, action, numéro = request.path.split('/')
# 根据其路径，参数 [numéro] 是一个正整数或零
numéro = int(numéro)
# 模拟 id=编号 必须存在于模拟列表中
simulations = session.get("simulations", [])
liste_simulations = list(filter(lambda simulation: simulation['id'] == numéro, simulations))
if not liste_simulations:
msg_erreur = f"la simulation n° [{numéro}] n'existe pas"
# 返回错误
return {"action": action, "état": 601, "réponse": [msg_erreur]}, status.HTTP_400_BAD_REQUEST
# 删除模拟 id=编号
simulation = liste_simulations.pop(0)
simulations.remove(simulation)
# 将模拟重新添加到会话中
session["simulations"] = simulations
# 返回结果
return {"action": action, "état": 600, "réponse": simulations}, status.HTTP_200_OK
- 第 10 行：获取请求路径中的两个元素。将其作为字符串获取；
- 第13行：将参数[numéro]转换为整数。由于其路由签名，我们知道这是可行的，
@app.route('/supprimer-simulation/<int:numero>', methods=['GET'])
此外，我们知道这是一个 >=0 的整数。实际上不可能存在 URL 或 [/supprimer-simulation/-4] 这样的参数。Flask 服务器会拒绝此类参数；
- 第 15 行：从会话中获取模拟列表；
- 第16行：使用函数[filter]，查找id==编号的模拟。得到一个[filter]对象，将其转换为[list]类型；
- 第17-20行：若过滤器未返回任何结果，则说明待删除的模拟不存在。返回一个指示此情况的错误响应；
- 第 21-23 行：删除过滤器返回的模拟；
- 第25行：将新的模拟列表重新加载到会话中；
- 第27行：在响应中返回新的模拟列表；
我们进行一次成功测试和一次失败测试。先执行模拟，然后请求模拟列表：
- 此处的模拟编号为2和3；
请求删除编号为3的模拟。
响应如下：
现在，让我们重新执行相同的操作（删除id=3的模拟）。此时的响应如下：
30.14. 操作 [fin-session]
操作 [fin-session] 允许用户结束其模拟会话。该操作的路径在脚本 [main] 中定义如下：
# 会话结束
@app.route('/fin-session', methods=['GET'])
def fin_session() -> tuple:
# 执行与该操作关联的控制器
return front_controller()
服务器不等待任何参数。该操作由以下 [FinSessionController] 控制器处理：
from flask_api import status
from werkzeug.local import LocalProxy
from InterfaceController import InterfaceController
class FinSessionController(InterfaceController):
def execute(self, request: LocalProxy, session: LocalProxy, config: dict) -> (dict, int):
# 获取路径中的元素
dummy, action = request.path.split('/')
# 删除当前会话中的所有键
session.clear()
# 返回结果
return {"action": action, "état": 400, "réponse": "session réinitialisée"}, status.HTTP_200_OK
- 第 13 行：删除会话中的所有键。这将删除：
- [typeResponse]：HTTP响应的类型（json、xml、html）；
- [id_simulation]：最近一次模拟的编号；
- [simulations]：用户的模拟列表；
- [user]：用户已通过身份验证的标志；
- 返回响应；
现在响应类型已不在会话中，可能会有人疑问第15行的响应HTTP将如何返回。 要了解这一点，需要回到主脚本 [main] 中的 |front_controller| 函数，并按以下方式进行修改：
…
# on not# 若会话中包含该信息，则记录所需的响应类型
type_response1 = session.get('typeResponse', None)
# 将请求转发至主控制器
main_controller = config['controllers']["main-controller"]
résultat, status_code = main_controller.execute(request, session, config)
# 记录发送给客户端的结果
log = f"[front_controller] {résultat}\n"
logger.write(log)
# 是否发生致命错误？
if status_code == status.HTTP_500_INTERNAL_SERVER_ERROR:
# 向应用程序管理员发送邮件
send_adminmail(config, log)
# 确定所需的响应类型
type_response2=session.get('typeResponse')
if type_response2 is None and type_response1 is None:
# 会话类型尚未确定——将采用 jSON
type_response = 'json'
elif type_response2 is not None:
# 已知响应类型且该类型已存在于会话中
type_response = type_response2
else:
type_response=type_response1
# 正在构建待发送的响应
response_builder = config["responses"][type_response]
response, status_code = response_builder \
.build_http_response(request, session, config, status_code, résultat)
# 发送响应
return response, status_code
- 第 3 行：存储当前会话中的响应类型；
- 第 6 行：执行该操作。如果操作为：
- [fin-session]，则键 [typeResponse] 将不再存在于会话中；
- [init-session]，则会话中的键 [typeResponse] 的值可能已发生变化；
- 第14-20行：必须发送响应HTTP。我们需要确定其具体形式：
- 第16-18行：如果响应类型既未由第3行的[type_response1]定义，也未由第15行的[type_response2]定义，则该响应类型在操作前后均未被定义。 此时使用 jSON（第 18 行）；
- 第19-21行：若存在[type_response2]（即操作后的会话响应类型），则应使用该类型；
- 第22-23行：否则则使用[type_response1]，即操作前的响应类型（该类型必然为[fin-session]）；
30.15. 操作 [get-admindata]
现在我们来讨论两个专用于服务 jSON 和 XML 的 URL：
操作 | 角色 | 执行上下文 |
/get-admindata | 返回用于计算税款的税务数据 | 请求 GET。 仅在会话类型为 json 或 xml 时使用。用户必须经过身份验证 |
/calculer-impots | 根据发布在 jSON 中的纳税人列表计算税款 | 请求 GET。 仅在会话类型为 json 或 xml 时使用。用户必须经过身份验证 |
URL [/get-admindata] 在主脚本 [main] 的路由中定义如下：
# 获取管理员数据
@app.route('/get-admindata', methods=['GET'])
def get_admindata() -> tuple:
# 执行与该操作关联的控制器
return front_controller()
路由 [/get-admindata] 由以下控制器 [GetAdminDataController] 处理：
# 导入依赖项
from flask_api import status
from werkzeug.local import LocalProxy
from InterfaceController import InterfaceController
class GetAdminDataController(InterfaceController):
def execute(self, request: LocalProxy, session: LocalProxy, config: dict) -> (dict, int):
# 获取路径中的元素
dummy, action = request.path.split('/')
# 仅接受 json 和 xml 格式的会话
type_response = session.get('typeResponse')
if type_response != 'json' and type_response != 'xml':
# 返回错误响应
return {
"action": action,
"état": 1001,
"réponse": ["cette action n'est possible que pour les sessions json ou xml"]
}, status.HTTP_400_BAD_REQUEST
else:
# 返回成功响应
return {"action": action, "état": 1000, "réponse": config["adminData"].asdict()}, status.HTTP_200_OK
- 第 13-21 行：验证当前是否处于 JSON 或 XML 会话中；
- 第24行：返回税务管理数据字典，该字典在服务器启动时已放置在配置中：
# admindata 将作为应用程序范围内的只读数据
config["admindata"] = config["layers"]["dao"].get_admindata()
我们使用Postman客户端，在启动jSON会话并完成身份验证后，请求URL和[/get-admindata]：
服务器的响应如下：
30.16. 操作 [calculer-impots]
操作 [calculer-impots] 用于计算请求正文中以字符串形式呈现的纳税人列表的税款，该字符串为 jSON。我们已知该操作：在上一版本中，它被称为 [calculate_tax_in_bulk_mode]。
其路径如下：
# 批量计算税款
@app.route('/calculer-impots', methods=['POST'])
def calculer_impots():
# 执行与该操作关联的控制器
return front_controller()
该操作由以下控制器 [CalculerImpotsController] 处理：
import json
from flask_api import status
from werkzeug.local import LocalProxy
from ImpôtsError import ImpôtsError
from InterfaceController import InterfaceController
from TaxPayer import TaxPayer
class CalculerImpotsController(InterfaceController):
def execute(self, request: LocalProxy, session: LocalProxy, config: dict) -> (dict, int):
# 获取路径中的元素
dummy, action = request.path.split('/')
# 仅接受 json 和 xml 格式的会话
type_response = session.get('typeResponse')
if type_response != 'json' and type_response != 'xml':
# 返回错误响应
return {
"action": action,
"état": 1501,
"réponse": ["cette action n'est possible que pour les sessions json ou xml"]
}, status.HTTP_400_BAD_REQUEST
# 获取 POST 请求正文 - 预期接收字典列表
msg_erreur = None
list_dict_taxpayers = None
# POST 的正文 jSON
request_text = request.data
try:
# 将其转换为字典列表
list_dict_taxpayers = json.loads(request_text)
except BaseException as erreur:
# 记录错误
msg_erreur = f"le corps du POST n'est pas une chaîne jSON valide : {erreur}"
# 列表是否不为空？
if not msg_erreur and (not isinstance(list_dict_taxpayers, list) or len(list_dict_taxpayers) == 0):
# 记录错误
msg_erreur = "le corps du POST n'est pas une liste ou alors cette liste est vide"
# 是否有一个字典列表？
if not msg_erreur:
erreur = False
i = 0
while not erreur and i < len(list_dict_taxpayers):
erreur = not isinstance(list_dict_taxpayers[i], dict)
i += 1
# 错误？
if erreur:
msg_erreur = "le corps du POST doit être une liste de dictionnaires"
# 错误？
if msg_erreur:
# 向客户端发送错误响应
résultats = {"action": action, "état": 1501, "réponse": [msg_erreur]}
return résultats, status.HTTP_400_BAD_REQUEST
# 逐个检查 TaxPayers
# 起初没有错误
list_erreurs = []
for dict_taxpayer in list_dict_taxpayers:
# 根据 dict_taxpayer 生成 TaxPayer
msg_erreur = None
try:
# 接下来的操作将排除参数不匹配的情况
# TaxPayer 类的属性，以及其值
# 值不正确的情况
TaxPayer().fromdict(dict_taxpayer)
except BaseException as erreur:
msg_erreur = f"{erreur}"
# 字典中必须包含某些键
if not msg_erreur:
# 密钥 [marié, enfants, salaire] 必须存在于字典中
keys = dict_taxpayer.keys()
if 'marié' not in keys or 'enfants' not in keys or 'salaire' not in keys:
msg_erreur = "le dictionnaire doit inclure les clés [marié, enfants, salaire]"
# 有错误吗？
if msg_erreur:
# 在 TaxPayer 本身中发现了错误
dict_taxpayer['erreur'] = msg_erreur
# 将 TaxPayer 添加到错误列表中
list_erreurs.append(dict_taxpayer)
# 已处理所有纳税人——是否有错误？
if list_erreurs:
# 向客户发送错误回复
résultats = {"action": action, "état": 1501, "réponse": list_erreurs}
return résultats, status.HTTP_400_BAD_REQUEST
# 无错误，可以继续处理
# 从税务机关获取数据
admindata = config["admindata"]
métier = config["layers"]["métier"]
try:
# 逐个处理 TaxPayer
list_taxpayers = []
for dict_taxpayer in list_dict_taxpayers:
# 计算税款
taxpayer = TaxPayer().fromdict(
{'marié': dict_taxpayer['marié'], 'enfants': dict_taxpayer['enfants'],
'工资：dict_taxpayer['salaire']})
métier.calculate_tax(taxpayer, admindata)
# 将结果保存为字典
list_taxpayers.append(taxpayer.asdict())
# 将 list_taxpayers 添加到当前的模拟中，并为每个模拟分配一个编号
simulations = session.get("simulations", [])
id_simulation = session.get("id_simulation", 0)
for simulation in list_taxpayers:
# 为每个模拟分配一个编号
id_simulation += 1
simulation['id'] = id_simulation
# 将其添加到当前的模拟列表中
simulations.append(simulation)
# 将所有内容重新加载到会话中
session["simulations"] = simulations
session["id_simulation"] = id_simulation
# 将响应发送给客户端
return {"action": action, "état": 1500, "réponse": list_taxpayers}, status.HTTP_200_OK
except ImpôtsError as erreur:
# 向客户端发送错误响应
return {"action": action, "état": 1501, "réponse": [f"{erreur}"]}, status.HTTP_500_INTERNAL_SERVER_ERROR
- 第 16-24 行：验证当前是否处于 JSON 或 XML 会话中
- 第26-120行：这段代码我们大致了解。它是应用程序第10版中 |index_controller| 函数的代码，经过调整以符合已实现的 [InterfaceController] 接口规范；
- 第104-115行：为适应该控制器的全新环境而添加的代码。我们刚刚完成了税款计算，需要将结果存储在会话中维护的模拟列表中；
- 第 105 行：获取当前会话中的模拟列表；
- 第 106 行：获取最近一次模拟的编号；
- 第107-112行：遍历税额计算结果字典列表，为每个字典分配一个[id]模拟编号，并将每个字典添加到模拟列表中；
- 第113-115行：将新的模拟列表以及上次执行的模拟编号重新写入会话；
在初始化 jSON 会话并完成身份验证后，我们执行以下 Postman 测试：
服务器的响应如下：
现在，如果查询模拟列表：
可以发现，在 [/calcul-impots] 的结果列表中，纳税人没有 [id] 属性，而在模拟列表中，每个模拟都有一个用于标识它的编号。
