Skip to content

23. 应用练习:第6版

23.1. 简介

现在我们回到税款计算应用程序。我们将以此为基础构建不同的Web应用程序。

在应用练习的第 5 版中,税务部门的数据存储在数据库中。该第 5 版包含两个独立但具有共同层级的应用程序:

  • 一个以 |批处理| 模式为存储在文本文件中的纳税人计算税款的应用程序;
  • 一个通过键盘输入纳税人信息,以 |交互| 模式计算税款的应用程序;

批处理(batch)模式的税款计算应用程序第5版具有以下架构:

Image

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

Image

  • 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

Image

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

Image

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

应用程序的架构如下:

Image

  • 浏览器 [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] 中:

Image

脚本 [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:

Image

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

Image

Image

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

Image


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

Image

服务器第 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

Image

版本 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 控制台中的客户端/服务器对话如下:

GET / HTTP/1.1
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: e65e2a28-4fe3-423b-88b3-b3e5a83092b1
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive

HTTP/1.0 401 UNAUTHORIZED
Content-Type: text/html; charset=utf-8
Content-Length: 19
WWW-Authenticate: Basic realm="Authentication Required"
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, 17 Jul 2020 07:05:37 GMT

Unauthorized Access
  • 第 10 行:服务器响应称我们无权访问 URL 和 [/]
  • 第13行:提示应使用的身份验证协议,此处为所谓的“基本身份验证”协议;

可以配置 Postman 使其根据 Basic 认证协议发送用户凭据:

Image

  • [6-7] 中,我们填入脚本 [config] 中的凭据: Image

    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 /:

Image

  • 与 Postman 一样,Firefox 从服务器收到了响应 HTTP,其头部信息为 HTTP:
1
2
3
4
HTTP/1.0 401 UNAUTHORIZED
WWW-Authenticate: Basic realm="Authentication Required"

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

Image

23.3. 税务计算服务器的Web客户端

23.3.1. 简介

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

Image

  • 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客户端脚本的目录结构如下:

Image

该脚本将实现自 |版本 1| 起描述的批处理模式税款计算应用程序。该应用程序的最新版本为 |版本 5|。其工作原理如下:

  • 待计算税款的纳税人被汇总在文本文件 [taxpayersdata.txt] 中:
# 有效数据:id、已婚、子女、工资
1,oui,2,55555
2,oui,2,50000
3,oui,3,50000
4,non,2,100000
5,non,3,100000
6,oui,3,100000
7,oui,5,100000
8,non,0,100000
9,oui,2,30000
10,non,0,200000
11,oui,3,200000
# 创建错误的记录
# 值不足
11,12
# 值有误
x,x,x,x
  • 计算结果将输出到两个文件中:
  • 文本文件 [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客户端配置

Image

配置通过两个脚本完成:

  • [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] 层的实现

Image

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

Image

  • [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&param2=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] 层;

该测试类将在以下环境中执行:

Image

  • 配置 [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 行:执行测试;

要执行测试,需创建一个 |执行配置|:

Image

  • 创建的是用于控制台脚本的运行配置,而非针对测试UnitTest的配置;

执行此配置时,将获得以下结果:

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-clients/01/tests/TestHttpClientDao.py
tests en cours...
{"réponse": {"result": {"marié": "oui", "enfants": 2, "salaire": 55555, "impôt": 2814, "surcôte": 0, "taux": 0.14, "décôte": 0, "réduction": 0}}}
....{"réponse": {"result": {"marié": "non", "enfants": 0, "salaire": 200000, "impôt": 64210, "surcôte": 7498, "taux": 0.45, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 3, "salaire": 200000, "impôt": 42842, "surcôte": 17283, "taux": 0.41, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 2, "salaire": 50000, "impôt": 1384, "surcôte": 0, "taux": 0.14, "décôte": 384, "réduction": 347}}}
{"réponse": {"result": {"marié": "oui", "enfants": 3, "salaire": 50000, "impôt": 0, "surcôte": 0, "taux": 0.14, "décôte": 720, "réduction": 0}}}
...{"réponse": {"result": {"marié": "non", "enfants": 2, "salaire": 100000, "impôt": 19884, "surcôte": 4480, "taux": 0.41, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "non", "enfants": 3, "salaire": 100000, "impôt": 16782, "surcôte": 7176, "taux": 0.41, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 3, "salaire": 100000, "impôt": 9200, "surcôte": 2180, "taux": 0.3, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 5, "salaire": 100000, "impôt": 4230, "surcôte": 0, "taux": 0.14, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "non", "enfants": 0, "salaire": 100000, "impôt": 22986, "surcôte": 0, "taux": 0.41, "décôte": 0, "réduction": 0}}}
....
{"réponse": {"result": {"marié": "oui", "enfants": 2, "salaire": 30000, "impôt": 0, "surcôte": 0, "taux": 0.0, "décôte": 0, "réduction": 0}}}
----------------------------------------------------------------------
Ran 11 tests in 0.130s

OK

Process finished with exit code 0

11 个测试均已通过。