23. 应用练习:第6版
23.1. 简介
现在我们回到税款计算应用程序。我们将以此为基础构建不同的Web应用程序。
在应用练习的第 5 版中,税务部门的数据存储在数据库中。该第 5 版包含两个独立但具有共同层级的应用程序:
- 一个以 |批处理| 模式为存储在文本文件中的纳税人计算税款的应用程序;
- 一个通过键盘输入纳税人信息,以 |交互| 模式计算税款的应用程序;
批处理(batch)模式的税款计算应用程序第5版具有以下架构:

最终,该应用程序的Web版本将采用以下架构:

- Web客户端 [1] 连接至 Web 服务器 [2],该服务器通过 SGBD 与 [3] 进行通信;
- Web 服务器 [2] 保留了初始应用程序中的 [métier]、[8] 以及 [dao]、[9] 层;
- 初始应用程序保留其主脚本 [4] 及其层 [métier] 和 [15]。 [métier]、[8] 和 [15] 层完全相同;
- 客户端/服务器通信需要两个额外层:
- 实现 Web 应用程序的 [web] [7] 层;
- 作为 Web 应用程序 [7] 客户端的 [dao] [5] 层;
在最终版本中,批量税款计算可通过两种方式进行:
- 由服务端的 [métier] 层执行业务税费计算。脚本 [main] 将采用此方法;
- 税款业务计算由客户端的 [métier] 层完成。脚本 [main2] 将采用此方法;
从现在开始,我们将开发多个上述类型的客户端/服务器应用程序,每个应用程序都将展示一种或多种新的Web开发技术。
23.2. 税款计算Web服务器
23.2.1. 版本 1

脚本 [server_01] 是以下 Web 应用程序:

- 在 [1] 中,使用了一个已配置的 URL,其中传递三个值:
- [marié](是/否)用于指示纳税人是否已婚;
- [enfants]:纳税人的子女数量;
- [salaire]:纳税人的年薪;
- 在 [2] 中,Web 服务器返回字符串 jSON,该字符串给出了应缴税额及其各项组成部分;
应用程序的架构如下:

- 浏览器 [1] 向服务器 [2] 发起请求。脚本 [server_01] 实现了服务器中的 [web] 和 [2] 层;
- [3-8] 层是税款计算应用程序 |第 5 版| 中已使用的层。我们直接沿用这些层;
- [métier] [3] 层在此处定义;
- [dao] [4] 层在此处定义;
Web 应用程序 [server_01] 通过三个脚本进行配置:
- [config] 用于配置整个应用程序;
- [config_database] 用于配置数据库访问。我们将使用 SGBD、MySQL 和 PostgreSQL;
- [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",
# IndexController
f"{script_dir}/../controllers",
# 脚本 [config_database, config_layers]
script_dir,
]
# 设置系统路径
from myutils import set_syspath
set_syspath(absolute_dependencies)
# 步骤 2 ------
# 应用程序配置
# 允许使用该应用程序的用户列表
config['users'] = [
{
"login": "admin",
"password": "admin"
}
]
# 步骤 3 ------
# 数据库配置
import config_database
config["database"] = config_database.configure(config)
# 步骤 4 ------
# 应用程序层实例化
import config_layers
config['layers'] = config_layers.configure(config)
# 应用配置
return config
- 函数 [configure] 接收字典 [config] 作为参数(第 1 行),并在丰富其内容后将其作为结果返回(第 54 行)。 其实很早就应该指出,没有必要将结果 [config] 返回。因为 [config] 是一个字典引用,调用方和被调用方共享该引用。 因此,调用方代码已经拥有该引用(第1行),无需再次返回(第54行)。因此,应改写为:
config=[module].configure(config) (1)
是多余的。只需写成:
[module].configure(config) (2)
不过,我还是保留了第一种写法,因为我觉得这样可能更能说明被调用代码修改了字典 [config]。
- 第 1 行:由函数 [configure] 接收的字典 [config] 包含一个名为“sgbd”的键,其值取自列表 [‘mysql’, ‘pgres’]。 [mysql] 表示所用数据库由 MySQL 管理,而 ‘pgres’ 表示所用数据库由 PostgreSQL 管理;
- 第 4-27 行:列出了所有包含 Web 应用程序所需组件的文件夹。这些文件夹将作为应用程序的 Python 路径(第 30-31 行)的一部分;
- 第 33-40 行:仅允许特定用户访问该应用程序。此处列出的用户列表中仅包含一名用户;
- 第 43-46 行:由脚本 [config_database] 构建所用数据库的配置;
- 第 46 行:由脚本 [config_database] 生成的配置是一个字典,将其存储在全局配置中,并关联到键 ‘database’;
- 第 48-51 行:脚本 [config_layers] 实例化 Web 应用程序的图层。它返回一个字典,该字典被存储在与键“layers”关联的通用配置中;
脚本 [config_database] 是 |第 5 版| 中已使用的脚本。为便于记忆,在此再次列出:
def configure(config: dict) -> dict:
# SQLAlchemy配置
from sqlalchemy import create_engine, Table, Column, Integer, MetaData, Float
from sqlalchemy.orm import mapper, sessionmaker
# 所用数据库的连接字符串
connection_strings = {
'mysql': "mysql+mysqlconnector://admimpots:mdpimpots@localhost/dbimpots-2019",
'pgres': "postgresql+psycopg2://admimpots:mdpimpots@localhost/dbimpots-2019"
}
# 所用数据库的连接字符串
engine = create_engine(connection_strings[config['sgbd']])
# 元数据
metadata = MetaData()
# 常量表
constantes_table = Table("tbconstantes", metadata,
Column('id', Integer, primary_key=True),
Column('plafond_qf_demi_part', Float, nullable=False),
Column('plafond_revenus_celibataire_pour_reduction', Float, nullable=False),
Column('plafond_revenus_couple_pour_reduction', Float, nullable=False),
Column('valeur_reduc_demi_part', Float, nullable=False),
Column('plafond_decote_celibataire', Float, nullable=False),
Column('plafond_decote_couple', Float, nullable=False),
Column('plafond_impot_celibataire_pour_decote', Float, nullable=False),
Column('plafond_impot_couple_pour_decote', Float, nullable=False),
Column('abattement_dixpourcent_max', Float, nullable=False),
Column('abattement_dixpourcent_min', Float, nullable=False)
)
# 税率区间表
tranches_table = Table("tbtranches", metadata,
Column('id', Integer, primary_key=True),
Column('limite', Float, nullable=False),
Column('coeffr', Float, nullable=False),
Column('coeffn', Float, nullable=False)
)
# 映射
from Tranche import Tranche
mapper(Tranche, tranches_table)
from Constantes import Constantes
mapper(Constantes, constantes_table)
# 会话工厂
session_factory = sessionmaker()
session_factory.configure(bind=engine)
# 一个会话
session = session_factory()
# 将某些信息保存并存储在字典中
return {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
"constantes_table": constantes_table, "session": session}
脚本 [config_layers] 用于配置 Web 服务器的层。这里再次使用一个已见过的 |脚本|:
def configure(config: dict) -> dict:
# 应用层的实例化
# DAO
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
dao = ImpotsDaoWithAdminDataInDatabase(config)
# 业务层
from ImpôtsMétier import ImpôtsMétier
métier = ImpôtsMétier()
# 将各层的实例放入字典中,并返回给调用方
return {
"dao": dao,
"métier": métier
}
- 第 6 行:[dao] 层通过数据库实现;
- [ImpotsDaoWithAdminDataInDatabase] 已在此处定义;
- [ImpôtsMétier] 已在此处定义;
主脚本 [server_01] 如下:
# 等待 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 ImpôtsError import ImpôtsError
from TaxPayer import TaxPayer
import re
from flask import request
from myutils import json_response
from flask import Flask
from flask_api import status
# 从税务管理部门获取数据
try:
# admindata 将作为应用程序范围内的只读数据
admindata = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
print(f"L'erreur suivante s'est produite : {erreur}")
sys.exit(1)
# Flask 应用程序
app = Flask(__name__)
# 首页 URL : /?已婚=xx&子女=yy&工资=zz
@app.route('/', methods=['GET'])
def index():
# 起初没有错误
erreurs = []
# URL 中的请求必须包含三个参数
if len(request.args) != 3:
erreurs.append("Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]")
# 从 URL 中获取婚姻状况
marié = request.args.get('marié')
if marié is None:
erreurs.append("paramètre [marié] manquant")
else:
marié = marié.strip().lower()
erreur = marié != "oui" and marié != "non"
if erreur:
erreurs.append(f"paramétre marié [{marié}] invalide")
# 从 URL 中获取子女数量
enfants = request.args.get('enfants')
if enfants is None:
erreurs.append("paramètre [enfants] manquant")
else:
enfants = enfants.strip()
match = re.match(r"^\d+", enfants)
if not match:
erreurs.append(f"paramétre enfants [{enfants}] invalide")
else:
enfants = int(enfants)
# 从 URL 中获取工资
salaire = request.args.get('salaire')
if salaire is None:
erreurs.append("paramètre [salaire] manquant")
else:
salaire = salaire.strip()
match = re.match(r"^\d+", salaire)
if not match:
erreurs.append(f"paramétre salaire [{salaire}] invalide")
else:
salaire = int(salaire)
# URL中存在无效参数?
for key in request.args.keys():
if key not in ['marié', 'enfants', 'salaire']:
erreurs.append(f"paramètre [{key}] invalide")
# 是否存在错误?
if erreurs:
# 向客户发送错误响应
résultats = {"réponse": {"erreurs": erreurs}}
return json_response(résultats, status.HTTP_400_BAD_REQUEST)
# 无错误,可以继续处理
# 计算税款
taxpayer = TaxPayer().fromdict({'marié': marié, 'enfants': enfants, 'salaire': salaire})
config["layers"]["métier"].calculate_tax(taxpayer, admindata)
# 向客户发送响应
return json_response({"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK)
# 仅主程序
if __name__ == '__main__':
# 启动 Flask 服务器
app.config.update(ENV="development", DEBUG=True)
app.run()
- 第 1-10 行:获取指示应使用哪个 SGBD 的参数;
- 第 12-14 行:利用该信息配置应用程序。特别是构建 Python 路径;
- 第 16-23 行:利用新的 Python 路径,导入所需的组件;
- 第25-31行:获取税务部门的数据,用于计算税款;
- 第 33-34 行:实例化 Flask 应用程序;
- 第38行:Flask应用程序仅处理URL [/]。它等待一个按以下方式配置的URL [/ ?marié=xx&enfants=yy&salaire=zz],其中:
- xx:是/否;
- yy:子女数量;
- zz:年薪;
- 第40-89行:验证URL参数的有效性;
- 第 41 行:将错误消息累积到列表 [erreurs] 中;
- 第43行:可能需要提醒的是,URL的配置参数可在[request.args]中找到(参见|此处|):
- 对象 [request] 是第 20 行导入的 Flask 对象;
- 对象 [request.args] 表现为字典;
- 第43-44行:验证参数数量恰好为三个(不多不少);
- 第46-49行:验证参数[marié]是否存在于URL中;
- 第 50-54 行:若存在,则验证其去除首尾空格后的纯文本值是否为“是”或“否”;
- 第56-59行:检查参数[enfants]是否存在于URL中;
- 第60-66行:如果存在该参数,则检查其值是否为正整数;
- 第 66 行:请注意,URL 的参数及其值均为字符串。参数 [enfants] 的值被转换为 'int' 类型;
- 第 68-78 行:对于参数 [salaire],执行与参数 [enfants] 相同的测试;
- 第 81-83 行:验证 URL 中是否仅包含 [‘marié, ‘enfants’, ‘salaire’] 这一参数;
- 第85-89行:若经过所有验证后,[erreurs]列表不为空,则将该错误列表以字符串jSON的形式连同状态码[400 Bad Request]发送给客户端;
由于后续我们经常需要向客户端发送字符串 jSON 作为响应,因此实现此发送所需的几行代码已被提取到我们之前用过的模块 [myutils.py] 中:

脚本 [myutils.py] 变为如下形式:
# 导入
import json
import os
import sys
from flask import make_response
def set_syspath(absolute_dependencies: list):
# absolute_dependencies:文件夹绝对路径列表
….
# 生成响应 HTTP jSON
def json_response(réponse: dict, status_code: int) -> tuple:
# 响应正文 HTTP
response = make_response(json.dumps(réponse, ensure_ascii=False))
# 响应正文 HTTP 来自 jSON
response.headers['Content-Type'] = 'application/json; charset=utf-8'
# 发送响应 HTTP
return response, status_code
- 第16行:函数[json_response]需要两个参数:
- [réponse]:需将字符串 jSON 发送给 Web 客户端的字典;
- [status_code]:响应的状态码 HTTP;
- 第 18 行:设置响应正文 jSON;
- 第 20 行:添加标头 HTTP,告知 Web 客户端将收到 jSON;
- 第 22 行:将响应 HTTP 发送给调用方代码。由其负责将其发送给 Web 客户端;
文件 [__init__.py] 的演变过程如下:
from .myutils import set_syspath, json_response
在 PyCharm 终端中,使用命令 [pip install .] 将 [myutils] 的新版本安装到机器范围模块中:
(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\packages>pip install .
Processing c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\packages
Using legacy setup.py install for myutils, since package 'wheel' is not installed.
Installing collected packages: myutils
Attempting uninstall: myutils
Found existing installation: myutils 0.1
Uninstalling myutils-0.1:
Successfully uninstalled myutils-0.1
Running setup.py install for myutils ... done
Successfully installed myutils-0.1
- 第 1 行:必须位于 [packages] 文件夹中才能输入此指令;
脚本 [server_01] 的代码后续如下:
…
# 有错误吗?
if erreurs:
# 向客户端发送错误响应
résultats = {"réponse": {"erreurs": erreurs}}
return json_response(résultats, status.HTTP_400_BAD_REQUEST)
# 无错误,可以继续处理
# 计算税款
taxpayer = TaxPayer().fromdict({'id': 0, 'marié': marié, 'enfants': enfants, 'salaire': salaire})
config["layers"]["métier"].calculate_tax(taxpayer, admindata)
# 向客户发送回复
return json_response({"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK)
- 第 10 行:此时,URL 中所需的参数已存在且正确;
- 第 10 行:创建对象 [TaxPayer] 以建模纳税人;
- 第 11 行:请求 [métier] 层计算税款。需注意,由 [métier] 层计算的元素会被插入作为参数传递的 [taxpayer] 对象中;
- 第13行:将响应以字符串jSON的形式发送给Web客户端。该字符串是字典中jSON的字符串。 将其与键 [result] 关联,并放入对象 [taxpayer] 的字典中。 无法直接将对象 [taxpayer] 本身放入,因为该对象无法序列化为 jSON;
创建两个运行配置,一个用于 MySQL,另一个用于 PostgreSQL:

以下是几个运行示例(您已启动应用程序 [server_01] 和 SGBD,随后通过浏览器请求 URL http://localhost:5000/):


以下是在 Postman 控制台中执行的示例:

GET /?mari%C3%A9=xx&enfants=yy&salaire=zz HTTP/1.1
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: e4c5df8c-4bd6-4250-b789-b7b164db4eff
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive
HTTP/1.0 400 BAD REQUEST
Content-Type: application/json; charset=utf-8
Content-Length: 134
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, 17 Jul 2020 06:15:44 GMT
{"réponse": {"erreurs": ["paramètre marié [xx] invalide", "paramètre enfants [yy] invalide", "paramètre salaire [zz] invalide"]}}
- 第 1 行:请求的 URL 格式不正确;
- 第 10 行:服务器返回 400 状态码 BAD REQUEST;
23.2.2. 版本 2

服务器第 2 版将 URL 的处理隔离在 [index_controller] [5] 模块中:
# 导入依赖项
import re
from flask_api import status
from werkzeug.local import LocalProxy
# URL 参数设置:/?已婚=xx&子女=yy&工资=zz
def execute(request: LocalProxy, config: dict) -> tuple:
# 依赖项
from TaxPayer import TaxPayer
# 初始时无错误
erreurs = []
# 请求必须包含三个参数
if len(request.args) != 3:
erreurs.append("Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]")
# 获取 URL 的婚姻状况
marié = request.args.get('marié')
if marié is None:
erreurs.append("paramètre [marié] manquant")
else:
marié = marié.strip().lower()
erreur = marié != "oui" and marié != "non"
if erreur:
erreurs.append(f"paramétre marié [{marié}] invalide")
# 获取 URL 的子女数量
enfants = request.args.get('enfants')
if enfants is None:
erreurs.append("paramètre [enfants] manquant")
else:
enfants = enfants.strip()
match = re.match(r"^\d+", enfants)
if not match:
erreurs.append(f"paramétre enfants {enfants} invalide")
else:
enfants = int(enfants)
# 获取 URL 的工资
salaire = request.args.get('salaire')
if salaire is None:
erreurs.append("paramètre [salaire] manquant")
else:
salaire = salaire.strip()
match = re.match(r"^\d+", salaire)
if not match:
erreurs.append(f"paramétre salaire {salaire} invalide")
else:
salaire = int(salaire)
# URL 中的其他参数?
for key in request.args.keys():
if not key in ['marié', 'enfants', 'salaire']:
erreurs.append(f"paramètre [{key}] invalide")
# 是否出现错误?
if erreurs:
# 向客户发送错误响应
résultats = {"réponse": {"erreurs": erreurs}}
return résultats, status.HTTP_400_BAD_REQUEST
# 无错误,可以继续处理
# 计算税款
taxpayer = TaxPayer().fromdict({'marié': marié, 'enfants': enfants, 'salaire': salaire})
config["layers"]["métier"].calculate_tax(taxpayer, config["admindata"])
# 向客户端发送响应
return {"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK
- 第 9 行:函数 [execute] 接收两个参数:
- [request]:来自客户端的 HTTP 请求;
- [config]:应用程序的配置字典;
脚本 [server_02] 如下:
# 等待 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 ImpôtsError import ImpôtsError
from flask import request
from myutils import json_response
from flask import Flask
import index_controller
# 从税务管理部门获取数据
try:
# admindata 将作为应用程序范围内的只读数据
config['admindata'] = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
print(f"L'erreur suivante s'est produite : {erreur}")
sys.exit(1)
# Flask 应用程序
app = Flask(__name__)
# 首页 URL : /?已婚=xx&子女=yy&工资=zz
@app.route('/', methods=['GET'])
def index():
# 执行请求
résultat, statusCode = index_controller.execute(request, config)
# 发送响应
return json_response(résultat, statusCode)
# 仅主程序
if __name__ == '__main__':
# 启动服务器
app.config.update(ENV="development", DEBUG=True)
app.run()
- 第 36-41 行:处理路径 /;
- 第 39 行:使用函数 [IndexController.execute];
今后我们将采用这种技术:每条路由都将由其专属的模块进行处理。
执行结果与版本 1 相同。
23.2.3. 版本 3

版本 3 引入了身份验证的概念。
脚本 [server_03] 变为如下:
# 等待 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 ImpôtsError import ImpôtsError
from flask import request
from myutils import json_response
from flask import Flask
from flask_httpauth import HTTPBasicAuth
import index_controller
# 从税务管理部门获取数据
try:
# config[‘admindata’] 将作为应用程序范围内的只读数据
config["admindata"] = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
print(f"L'erreur suivante s'est produite : {erreur}")
sys.exit(1)
# 身份验证管理器
auth = HTTPBasicAuth()
# 身份验证方法
@auth.verify_password
def verify_credentials(login: str, password: str) -> bool:
# 用户列表
users = config['users']
# 遍历此列表
for user in users:
if user['login'] == login and user['password'] == password:
return True
# 未找到
return False
# Flask 应用程序
app = Flask(__name__)
# 首页 URL : /?已婚=xx&子女=yy&工资=zz
@app.route('/', methods=['GET'])
@auth.login_required
def index():
# 执行请求
résultat, statusCode = index_controller.execute(request, config)
# 发送响应
return json_response(résultat, statusCode)
# 仅限主程序
if __name__ == '__main__':
# 启动服务器
app.config.update(ENV="development", DEBUG=True)
app.run()
- 第 21 行:导入一个身份验证管理器。Web 服务器支持多种身份验证类型。此处使用的名为 [HTTP Basic]。每种身份验证类型都遵循特定的客户端/服务器交互流程;
- 第 33 行:创建身份验证管理器的实例;
- 第 37 行:注解 [@auth.verify_password] 标记了当身份验证管理器需要根据 [HTTP Basic] 协议验证客户端传来的用户名/密码时应执行的函数;
- 第 55 行:注解 [@auth.login_required] 标记了一条需要对 Web 客户端进行身份验证的路由。如果 Web 客户端尚未发送其凭据,Web 服务器将根据 HTTP 基本协议自动向其请求凭据;
必须安装模块 [flask_httpauth]:
(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-servers\01\flask>pip install flask_httpauth
Collecting flask_httpauth
Downloading Flask_HTTPAuth-4.1.0-py2.py3-none-any.whl (5.8 kB)
Requirement already satisfied: Flask in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from flask_httpauth) (1.1.2)
Requirement already satisfied: itsdangerous>=0.24 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (1.1.0)
Requirement already satisfied: click>=5.1 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (7.1.2)
Requirement already satisfied: Jinja2>=2.10.1 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (2.11.2)
Requirement already satisfied: Werkzeug>=0.15 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (1.0.1)
Requirement already satisfied: MarkupSafe>=0.23 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Jinja2>=2.10.1->Flask->flask_httpauth) (1.1.1
)
Installing collected packages: flask-httpauth
Successfully installed flask-httpauth-4.1.0
让我们通过 Postman 控制台观察具体情况。您:
- 创建一个运行配置;
- 启动 Web 应用程序;
- 启动您选择的 SGBD;
- 使用 Postman 请求 URL 和 [/];
Postman 控制台中的客户端/服务器对话如下:
- 第 10 行:服务器响应称我们无权访问 URL 和 [/];
- 第13行:提示应使用的身份验证协议,此处为所谓的“基本身份验证”协议;
可以配置 Postman 使其根据 Basic 认证协议发送用户凭据:

- 在 [6-7] 中,我们填入脚本 [config] 中的凭据:
config['users'] = [
{
"login": "admin",
"password": "admin"
}
]
Postman 控制台中的客户端/服务器对话如下所示:
GET / HTTP/1.1
Authorization: Basic YWRtaW46YWRtaW4=
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: 5ce20822-e87c-4eef-a2f4-b9eaec38d881
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive
HTTP/1.0 400 BAD REQUEST
Content-Type: application/json; charset=utf-8
Content-Length: 203
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, 17 Jul 2020 07:20:01 GMT
{"réponse": {"erreurs": ["Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]", "paramètre [marié] manquant", "paramètre [enfants] manquant", "paramètre [salaire] manquant"]}}
- 第 2 行:Postman 客户端以加密形式发送用户 [admin / admin] 的凭据;
- 第 17 行:服务器响应正常。由于未发送参数 [marié, enfants, salaire](第 1 行),服务器报告了错误,但未报告身份验证错误;
现在我们使用浏览器(下图为 Firefox)请求 URL /:

- 与 Postman 一样,Firefox 从服务器收到了响应 HTTP,其头部信息为 HTTP:
Firefox与其他浏览器一样,在收到这些标头时不会终止对话。它会要求用户输入服务器所需的凭据。只需在上述界面输入 admin / admin 即可收到服务器的响应:

23.3. 税务计算服务器的Web客户端
23.3.1. 简介
在前一节中,税费计算服务器的Web客户端是一个浏览器。在本节中,Web客户端将是一个控制台脚本。架构如下:

- Web客户端由[1-2]层组成;
- Web 服务器由 [3-9] 层组成。上一节已对此进行了说明;
因此我们需要编写 [1-2] 层。
[dao] 和 [2] 层必须能够与 [3] 网络服务器进行通信。 现在我们已经了解了 HTTP 协议,因此可以利用之前学习过的 [pycurl] 模块,编写一个与 Web 服务器 [3] 通信的脚本。 不过,还有专门用于客户端/服务器对话的模块 HTTP。我们将使用其中之一,即模块 [requests]:
(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-servers\01\flask>pip install requests
Collecting requests
Downloading requests-2.24.0-py2.py3-none-any.whl (61 kB)
|| 61 kB 137 kB/s
Collecting idna<3,>=2.5
Downloading idna-2.10-py2.py3-none-any.whl (58 kB)
|| 58 kB 692 kB/s
Collecting chardet<4,>=3.0.2
Downloading chardet-3.0.4-py2.py3-none-any.whl (133 kB)
|| 133 kB 1.3 MB/s
Collecting urllib3!=1.25.0,!=1.25.1,<1.26,>=1.21.1
Downloading urllib3-1.25.9-py2.py3-none-any.whl (126 kB)
|| 126 kB 1.1 MB/s
Collecting certifi>=2017.4.17
Downloading certifi-2020.6.20-py2.py3-none-any.whl (156 kB)
|| 156 kB 1.1 MB/s
Installing collected packages: idna, chardet, urllib3, certifi, requests
Successfully installed certifi-2020.6.20 chardet-3.0.4 idna-2.10 requests-2.24.0 urllib3-1.25.9
Web客户端脚本的目录结构如下:

该脚本将实现自 |版本 1| 起描述的批处理模式税款计算应用程序。该应用程序的最新版本为 |版本 5|。其工作原理如下:
- 待计算税款的纳税人被汇总在文本文件 [taxpayersdata.txt] 中:
- 计算结果将输出到两个文件中:
- 文本文件 [errors.txt] 汇总纳税人文件中检测到的错误:
Analyse du fichier C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-clients\01\main/../data/input/taxpayersdata.txt
Ligne 15, not enough values to unpack (expected 4, got 2)
Ligne 17, MyException[1, L'identifiant d'une entité <class 'TaxPayer.TaxPayer'> doit être un entier >=0]
- (续)
- 文件 jSON [résultats.json] 汇总了各纳税人的税款计算结果:
[
{
"id": 0,
"marié": "oui",
"enfants": 2,
"salaire": 55555,
"impôt": 2814,
"surcôte": 0,
"taux": 0.14,
"décôte": 0,
"réduction": 0
},
{
"id": 1,
"marié": "oui",
"enfants": 2,
"salaire": 50000,
"impôt": 1384,
"surcôte": 0,
"taux": 0.14,
"décôte": 384,
"réduction": 347
},
…
]
23.3.2. Web客户端配置

配置通过两个脚本完成:
- [config],负责架构非分层部分的全部配置;
- [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",
# ImpôtsDaoWithHttpClient
f"{script_dir}/../services",
# 配置脚本
script_dir,
]
# 设置 syspath
from myutils import set_syspath
set_syspath(absolute_dependencies)
# 步骤 2------
# 使用常量配置应用程序
config.update({
"taxpayersFilename": f"{script_dir}/../data/input/taxpayersdata.txt",
"resultsFilename": f"{script_dir}/../data/output/résultats.json",
"errorsFilename": f"{script_dir}/../data/output/errors.txt",
"server": {
"urlServer": "http://127.0.0.1:5000/",
"authBasic": True,
"user": {
"login": "admin",
"password": "admin"
}
}
}
)
# 步骤 3 ------
# 实例化层
import config_layers
config['layers'] = config_layers.configure(config)
# 应用配置
return config
- 第 1 行:函数 [configure] 接收一个字典作为参数,该字典需填充配置信息。该字典可以是预先填充好的,也可以是空的。在此示例中,它将是空的;
- 第 40-42 行:由 [dao] 层管理的三个文本文件的绝对路径;
- 第43-50行:与键[server]相关联,[dao]层需要了解的、关于其需与其通信的Web服务器的信息:
- 第 44 行:Web 服务的 URL;
- 第 45 行:若访问 URL 需要 Basic 类型认证,则 [authBasic] 键的值为 True;
- 第46-49行:若需进行身份验证,此处为将进行身份验证的用户凭据;
- 第56-57行:实例化层,此处为唯一的[dao]层,并将层引用放入[config]中,该引用与密钥[layers]相关联;
脚本 [config_layers] 如下:
def configure(config: dict) -> dict:
# 实例化应用程序层
# DAO层
from ImpôtsDaoWithHttpClient import ImpôtsDaoWithHttpClient
dao = ImpôtsDaoWithHttpClient(config)
# 完成层配置
return {
"dao": dao
}
- 第1行:函数[configure]接收用于配置应用程序的字典;
- 第4-6行:实例化[dao]层。第6行,向其传递应用程序的配置,该配置中包含其所需的信息;
- 第8-11行:返回一个字典,其中包含[dao]层的引用;
23.3.3. 主脚本 [main]
主脚本 [main] 是 |第 5 版| 脚本的一个变体:
# 配置应用程序
import config
config = config.configure({})
# 依赖关系
from ImpôtsError import ImpôtsError
# 代码
try:
# 检索图层[dao]
dao = config["layers"]["dao"]
# 读取纳税人数据
taxpayers = dao.get_taxpayers_data()["taxpayers"]
# 纳税人?
if not taxpayers:
raise ImpôtsError(f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
# 计算纳税人的税款
for taxpayer in taxpayers:
# 纳税人既是输入参数也是输出参数
# 纳税人将被修改
dao.calculate_tax(taxpayer)
# 将结果写入文本文件
dao.write_taxpayers_results(taxpayers)
except ImpôtsError as erreur:
# 显示错误
print(f"L'erreur suivante s'est produite : {erreur}")
finally:
# 完成
print("Travail terminé...")
- 第2-3行:应用程序已配置;
- 第13行:[dao]层提供需计算税款的纳税人列表;
- 第 21 行:[dao] 层计算每位纳税人的税额;
- 第23行:结果被保存到文件jSON中;
23.3.4. [dao] 层的实现

让我们回顾一下所使用的客户端/服务器架构:

- 在 [2, 6] 中,我们可以看到 [dao] 层具有两个作用:
- 它既访问文件系统以读取纳税人的数据,也写入税款计算结果。我们已经有一个 |AbstractImpôtsDao| 类能够完成这些操作。该类自 |第 4 版| 起便已投入使用;
- 它与 Web 服务器 [3] 进行交互;
在 |第 5 版| 中,主脚本 [main] [1] 直接与 [métier] [4] 层进行交互。 我们希望不修改该脚本。为此,我们将确保 [dao] [2] 层实现 [métier] [4] 层的接口。 这样,主脚本 [main] 就会感觉像是在直接与 [métier] [4] 层进行交互,并完全忽略该层位于另一台机器上的事实。
实现 [dao] [2] 层的类定义可以如下所示:
class ImpôtsDaoWithHttpClient(AbstractImpôtsDao, InterfaceImpôtsMétier):
- 类 [ImpôtsDaoWithHttpClient]:
- 继承自类 [AbstractImpôtsDao],从而能够管理与文件系统 [6] 的交互;
- 实现了接口 [InterfaceImpôtsMétier],从而无需修改 |第 5 版| 的主脚本 [main];
[ImpôtsDaoWithHttpClient]类的完整代码如下:
# 导入
import requests
from flask_api import status
from AbstractImpôtsDao import AbstractImpôtsDao
from AdminData import AdminData
from ImpôtsError import ImpôtsError
from InterfaceImpôtsMétier import InterfaceImpôtsMétier
from TaxPayer import TaxPayer
class ImpôtsDaoWithHttpClient(AbstractImpôtsDao, InterfaceImpôtsMétier):
# 构造函数
def __init__(self, config: dict):
# 父类初始化
AbstractImpôtsDao.__init__(self, config)
# 参数存储
self.__config_server = config["server"]
# [AbstractImpôtsDao] 的未使用方法
def get_admindata(self) -> AdminData:
pass
# 税款计算
def calculate_tax(self: object, taxpayer: TaxPayer, admindata: AdminData = None):
# 允许异常上报
# GET 参数
params = {"marié": taxpayer.marié, "enfants": taxpayer.enfants, "salaire": taxpayer.salaire}
# 使用 Basic 认证进行连接?
if self.__config_server['authBasic']:
response = requests.get(
# 被查询服务器的 URL
self.__config_server['urlServer'],
# URL 的参数
params=params,
# Basic 身份验证
auth=(
self.__config_server["user"]["login"],
self.__config_server["user"]["password"]))
else:
# 不使用 Basic 身份验证的连接
response = requests.get(self.__config_server['urlServer'], params=params)
# 验证
print(response.text)
# 响应状态码 HTTP
status_code = response.status_code
# 将响应 jSON 放入字典
résultat = response.json()
# 如果状态码不为 200,则报错 OK
if status_code != status.HTTP_200_OK:
# 已知错误已关联至响应中的键 [erreurs]
raise ImpôtsError(87, résultat['réponse']['erreurs'])
# 已知该结果与响应中的键 [result] 相关联
# 使用该结果修改输入参数
taxpayer.fromdict(résultat["réponse"]["result"])
- 第 21-23 行:类 [AbstractImpôtsDao](第 12 行)包含一个抽象方法 [get_admindata]。即使我们不使用它(admindata 由服务器而非客户端管理),也必须对其进行实现;
- 第 26 行:方法 [calculate_tax] 属于接口 [InterfaceImpôtsMétier](第 12 行)。我们需要对其进行实现;
- 第 15 行:构造函数接收的唯一参数是应用程序配置字典;
- 第16-17行:通过同样传入应用程序配置来初始化父类[AbstractImpôtsDao]。该类将从中获取其需要管理的三个文本文件的名称;
- 第18-19行:在类中本地存储与税款计算Web服务器相关的信息;
- 第26行:方法 [calculate_tax] 接收一个类型为 |Taxpayer| 的对象作为参数。为了符合方法 [InterfaceImpôtsMétier.calculate_tax] 的签名,它还接收一个参数 [admindata],该参数旨在封装税务部门的数据。 客户端并不拥有这些数据。该参数始终保持为 [None]。这种扭曲的实现方式表明,[ImpôtsMétier] 类最初编写有误:
- [calculate_tax] 的签名本应仅为:
def calculate_tax(self, taxpayer: TaxPayer)
,而参数 [admindata : AdminData] 本应传递给该类的构造函数;
- 第 27 行:方法 [calculate_tax] 的代码未封装在 try / catch / finally 块中。这意味着可能发生的异常将无法被处理,并会回传至调用方代码,即脚本 [main]。 该脚本确实捕获了从 [dao] 层回传的所有异常;
- 第 28 行:税款计算在服务器端进行。因此需要与服务器进行通信。这通过第 2 行导入的 [requests] 模块实现;
- 第31-43行:为向Web服务器发送GET请求,使用[requests.get]方法:
- 第33-34行:该方法的第一个参数是要联系的URL;
- 第35-40行:其余两个参数是命名参数,其顺序无关紧要;
- 第35-36行:名为[params]的参数值必须是一个字典,其中包含以[/url ?param1=valeur1¶m2=valeur2&…]格式写入URL中的信息;
- 第29行:包含Web服务器所期待的三个参数[marié, enfants, salaire]的字典。无需处理这些参数必须经过的编码(称为urlencoded),[requests]会自动处理;
- 第 37-40 行:名为 [auth] 的参数是一个包含两个元素(login、password)的元组。它代表 Basic 类型身份验证的凭据;
- 第 44-45 行:这两行仅用于教学目的(调试完成后将注释掉):
- [response] 代表服务器的响应 HTTP;
- [response.text] 代表封装在此响应中的文档文本。在调试阶段,检查服务器发送的内容非常有用;
- 第 47 行:[response.status_code] 是接收到的响应的状态码 HTTP。我们的服务器仅发送三种状态码:
- 200 OK
- 400 BAD REQUEST
- 500 INTERNAL SERVER ERROR
- 第 49 行:即使出现错误,我们的服务器仍会发送 jSON。 函数 [response.json()] 根据接收到的字符串 jSON 创建一个字典。回顾字符串 jSON 的两种可能形式:
{"réponse": {"erreurs": ["Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]", "paramètre [marié] manquant", "paramètre [enfants] manquant", "paramètre [salaire] manquant"]}}
{"réponse": {"result": {"id": 0, "marié": "oui", "enfants": 3, "salaire": 200000, "impôt": 42842, "surcôte": 17283, "taux": 0.41, "décôte": 0, "réduction": 0}}}
- 第 51-53 行:如果状态码不是 200,则抛出异常,并将错误消息封装在响应中;
- 第 56 行:获取税额计算生成的字典,并将其用于更新输入参数 [taxpayer];
23.3.5. 执行
要运行客户端:
- 启动服务器 [server_03],并指定您选择的 SGBD;
- 运行客户端脚本 [main];
结果将保存在文件夹 [data/output] 中。这些结果与第 5 版相同。
23.4. [dao] 层的测试
让我们回到客户端/服务器应用程序的架构:
- 在客户端代码中,我们设法使 [dao] 和 [1] 层提供与 [métier] 和 [3] 层相同的接口。 因此,我们将使用 [4],并利用之前已研究的测试类 |TestDaoMétier| 来测试 [métier] [3] 层;
该测试类将在以下环境中执行:

- 配置 [2] 与我们刚刚研究的配置 [1] 完全相同;
测试类 [TestHttpClientDao] 如下:
import unittest
class TestHttpClientDao(unittest.TestCase):
def test_1(self) -> None:
from TaxPayer import TaxPayer
# {'已婚': '是', '子女': 2, '工资': 55555,
# '税款': 2814, '加征': 0, '减免': 0, '减免': 0, '税率': 0.14}
taxpayer = TaxPayer().fromdict({"marié": "oui", "enfants": 2, "salaire": 55555})
dao.calculate_tax(taxpayer)
# 核对
self.assertAlmostEqual(taxpayer.impôt, 2815, delta=1)
self.assertEqual(taxpayer.décôte, 0)
self.assertEqual(taxpayer.réduction, 0)
self.assertAlmostEqual(taxpayer.taux, 0.14, delta=0.01)
self.assertEqual(taxpayer.surcôte, 0)
…
def test_11(self) -> None:
from TaxPayer import TaxPayer
# {'已婚': '是', '子女': 3, '工资': 200000,
# '税款': 42842, '附加费': 17283, '减免': 0, '减免额': 0, '税率': 0.41}
taxpayer = TaxPayer().fromdict({'marié': 'oui', 'enfants': 3, 'salaire': 200000})
dao.calculate_tax(taxpayer)
# 核查
self.assertAlmostEqual(taxpayer.impôt, 42842, 1)
self.assertEqual(taxpayer.décôte, 0)
self.assertEqual(taxpayer.réduction, 0)
self.assertAlmostEqual(taxpayer.taux, 0.41, delta=0.01)
self.assertAlmostEqual(taxpayer.surcôte, 17283, delta=1)
if __name__ == '__main__':
# 配置应用程序
import config
config = config.configure({})
# DAO层
dao = config['layers']['dao']
# 执行测试方法
print("tests en cours...")
unittest.main()
该类与应用程序第 4 版中已研究的类类似。
- 第 40-41 行:配置测试环境;
- 第 44 行:获取 [dao] 层的引用;
- 第 47-48 行:执行测试;
要执行测试,需创建一个 |执行配置|:

- 创建的是用于控制台脚本的运行配置,而非针对测试UnitTest的配置;
执行此配置时,将获得以下结果:
11 个测试均已通过。