20. 应用练习：第 5 版

我们将开发三个应用程序：

• 应用程序 1 将初始化数据库，该数据库将取代第 4 版中的文件 [admindata.json]；
• 应用程序2将在批处理模式下进行税款计算；
• 应用程序 3 将以交互模式计算税款；

20.1. 应用程序 1：数据库初始化

应用程序 1 将采用以下架构：

这是对第4版架构（参见“第4版”章节）的改进：税务数据将存储在数据库中，而非jSON文件中。[dao]层将进行调整以实现这一变更。

20.1.1. 文件 [admindata.json]

文件 [admindata.json] 与第 4 版中的内容相同：

{
    "limites": [9964, 27519, 73779, 156244, 0],
    "coeffr": [0, 0.14, 0.3, 0.41, 0.45],
    "coeffn": [0, 1394.96, 5798, 13913.69, 20163.45],
    "plafond_qf_demi_part": 1551,
    "plafond_revenus_celibataire_pour_reduction": 21037,
    "plafond_revenus_couple_pour_reduction": 42074,
    "valeur_reduc_demi_part": 3797,
    "plafond_decote_celibataire": 1196,
    "plafond_decote_couple": 1970,
    "plafond_impot_couple_pour_decote": 2627,
    "plafond_impot_celibataire_pour_decote": 1595,
    "abattement_dixpourcent_max": 12502,
    "abattement_dixpourcent_min": 437
}

我们将使用该字典中的键作为数据库的列。

20.1.2. 创建数据库

正如在 |创建 MySQL 数据库| 段落中所示， 我们创建一个名为 [dbimpots-2019] 的数据库，其所有者为 [admimpots]，密码为 [mdpimpots]。 在 [phpMyAdmin] 中，结果如下：

同样，正如在“创建数据库 PostgreSQL”一节中所示， 我们创建了一个名为 PostgreSQL 的数据库，该数据库属于用户 [admimpots]，密码为 [mdpimpots]。 在 [pgAdmin] 中，结果如下：

数据库已创建，但目前尚无任何表。这些表将由 ORM 和 [sqlalchemy] 构建。

20.1.3. 由 [sqlalchemy] 映射的实体

我们将创建两个表来封装 [admindata.json] 的数据：

由 [sqlalchemy] 定义的表 [tbtranches] 将汇总来自字典 [admindata.json] 中表 [limites, coeffr, coeffn] 的数据：

    # 显示税率区间表
    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)
                           )

由 [sqlalchemy] 定义的表 [tbconstantes] 将汇总字典 [admindata.json] 中的常量：

    # 常量表
    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)
                             )

将与这两个表进行映射的实体如下：

实体 [Constantes] 封装了字典 [admindata.json] 中的常量：

from BaseEntity import BaseEntity


# 税务管理数据容器类
class Constantes(BaseEntity):
    # 类状态中的排除键
    excluded_keys = ["_sa_instance_state"]

    # 允许的键
    @staticmethod
    def get_allowed_keys() -> list:
        return ["id",
                "plafond_qf_demi_part",
                "plafond_revenus_celibataire_pour_reduction",
                "plafond_revenus_couple_pour_reduction",
                "valeur_reduc_demi_part",
                "plafond_decote_celibataire",
                "plafond_decote_couple",
                "plafond_decote_couple",
                "plafond_impot_celibataire_pour_decote",
                "plafond_impot_couple_pour_decote",
                "abattement_dixpourcent_max",
                "abattement_dixpourcent_min"]

• 第 5 行：类 [Constantes] 继承自类 [BaseEntity]；
• 第 7 行：通过映射 [sqlalchemy]，类 [Constante] 将获得属性 [_sa_instance_state]。我们将该属性从实体的字典 [asdict] 中移除；
• 第11-23行：实体的属性。为便于编写代码，我们沿用了[admindata.json]字典中使用的名称；

实体 [Tranche] 封装了字典 [admindata.json] 中三个数组 [limites, coeffr, coeffn] 的一行：

from BaseEntity import BaseEntity


# 税务管理数据容器类
class Tranche(BaseEntity):
    # 从类状态中排除的键
    excluded_keys = ["_sa_instance_state"]

    # 允许的键
    @staticmethod
    def get_allowed_keys() -> list:
        return ["id", "limite", "coeffr", "coeffn"]

• 第 5 行：类 [Tranche] 继承自类 [BaseEntity]；
• 第 7 行：从该实体的 [asdict] 字典属性中，排除由 [sqlalchemy] 添加的 [_sa_instance_state] 属性；
• 第10-12行：类的属性；

实体 [Constantes, Tranche] 与表 [constantes, tranches] 之间的映射如下：

…
    # 常量表
    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)

• 映射关系在第24-29行完成。其中省略了映射实体属性与数据库表之间的对应关系。当表的列名与应关联的属性名称相同时，这种做法是可行的。因此，我们在表中采用了映射实体的属性名称。这有助于编写代码并提高可读性；

20.1.4. [sqlalchemy]的配置文件

我们刚刚详细介绍了 [sqlalchemy] 配置文件的一部分。完整的 [config_database] 文件如下：

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()

    # 记录某些信息
    config['database'] = {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
                          "constantes_table": constantes_table, "session": session}

    # 结果
    return config

• 第1行：函数[configure]接收一个字典作为参数，其中键[sgbd]指定应使用哪个SGBD： MySQL（MySQL）或 PostgreSQL（PostgreSQL）；
• 第 6-12 行：根据配置要求选择数据库；
• 第14-44行：实体与表的映射。这些映射很简单，因为[tranches]表和[constantes]表之间没有任何关联。它们是独立的。因此，无需管理两者之间的外键关系；
• 第 46-51 行：创建应用程序的工作会话 [session]；
• 第53-58行：将相关信息写入配置字典，并返回该字典；

20.1.5. [dao]层

让我们回到待构建的应用程序 1 的架构：

[dao] [1] 层需读取 [admindata.json] [2] 文件，并将内容传输至 [3, 4] 数据库之一；

层 [dao] 实现了接口 [1]，并由类 [2] 实现。

接口 [InterfaceDao4TransferAdminData2Database] 如下所示：

# 导入
from abc import ABC, abstractmethod


# 接口 InterfaceImpôtsUI
class InterfaceDao4TransferAdminData2Database(ABC):
    # 将税务数据传输到数据库
    @abstractmethod
    def transfer_admindata_in_database(self:object):
        pass

• 第 8-10 行：该接口仅定义了一个无参数的方法 [transfer_admindata_in_database]。由于该方法需要参数（哪个文件？哪个数据库？），这意味着这些参数将传递给实现该接口的类的构造函数；

类 [DaoTransferAdminDataFromJsonFile2Database] 以如下方式实现了接口 [InterfaceDao4TransferAdminData2Database]：

# 导入
import codecs
import json

from sqlalchemy.exc import DatabaseError, IntegrityError, InterfaceError

from Constantes import Constantes
from ImpôtsError import ImpôtsError
from InterfaceDao4TransferAdminData2Database import InterfaceDao4TransferAdminData2Database
from Tranche import Tranche


class DaoTransferAdminDataFromJsonFile2Database(InterfaceDao4TransferAdminData2Database):

    # 生成器
    def __init__(self, config: dict):
        self.config = config

    # 传输
    def transfer_admindata_in_database(self) -> None:
        # 初始化
        session = None
        config = self.config

        try:
            # 从税务部门获取数据
            with codecs.open(config["admindataFilename"], "r", "utf8") as fd:
                # 将内容传输到字典
                admindata = json.load(fd)

            # 获取数据库配置
            database = config["database"]

            # 从数据库中删除两个表
            # checkfirst=True：首先检查表是否存在
            database["tranches_table"].drop(database["engine"], checkfirst=True)
            database["constantes_table"].drop(database["engine"], checkfirst=True)

            # 根据映射关系重建表
            database["metadata"].create_all(database["engine"])

            # 当前会话 [sqlalchemy]
            session = database["session"]

            # 填充税率区间表
            limites = admindata["limites"]
            coeffr = admindata["coeffr"]
            coeffn = admindata["coeffn"]
            for i in range(len(limites)):
                session.add(Tranche().fromdict(
                    {"limite": limites[i], "coeffr": coeffr[i], "coeffn": coeffn[i]}))
            # 填充常量表
            session.add(Constantes().fromdict({
                'plafond_qf_demi_part': admindata["plafond_qf_demi_part"],
                'plafond_revenus_celibataire_pour_reduction': admindata["plafond_revenus_celibataire_pour_reduction"],
                'plafond_revenus_couple_pour_reduction': admindata["plafond_revenus_couple_pour_reduction"],
                'valeur_reduc_demi_part': admindata["valeur_reduc_demi_part"],
                'plafond_decote_celibataire': admindata["plafond_decote_celibataire"],
                'plafond_decote_couple': admindata["plafond_decote_couple"],
                'plafond_impot_celibataire_pour_decote': admindata["plafond_impot_celibataire_pour_decote"],
                'plafond_impot_couple_pour_decote': admindata["plafond_impot_couple_pour_decote"],
                'abattement_dixpourcent_max': admindata["abattement_dixpourcent_max"],
                'abattement_dixpourcent_min': admindata["abattement_dixpourcent_min"]
            }))

            # 会话验证 [sqlalchemy]
            session.commit()
        except (IntegrityError, DatabaseError, InterfaceError) as erreur:
            # 以另一种形式重新抛出异常
            raise ImpôtsError(17, f"{erreur}")
        finally:
            # 释放会话资源
            if session:
                session.close()

• 第 13 行：类 [DaoTransferAdminDataFromJsonFile2Database] 实现了接口 [InterfaceDao4TransferAdminData2Database]；
• 第 15-17 行：该类的构造函数接收配置字典作为参数。将使用以下键：
  • [admindataFilename]（第27行）：包含需导入数据库的税务管理数据的文件jSON的名称；
  • [database] 第32行：应用程序的配置[sqlalchemy]；
• 第34-37行：若存在，则删除表[constantes]和[tranches]；
• 第 39-40 行：重新创建这两个表；
• 第 43 行：获取配置中存在的 [sqlalchemy] 会话；
• 第 45-51 行：将字典 [admindata] 中的表 [limites, coeffr, coeffn] 放入会话。为此，将实体 [Tranche] 的实例放入会话；
• 第 52-64 行：将 [Constantes] 实体的一个实例放入会话中；
• 第66-67行：会话被验证。如果会话数据尚未存入数据库，则此时将其存入；
• 第68-70行：处理可能出现的错误；
• 第 71-74 行：关闭会话。由于 [dao] 层仅使用一次，因此可以执行此操作；

20.1.6. 应用程序配置

该应用程序由三个文件进行配置：

• [config] 是通用配置文件。它负责配置应用程序 [main]。该文件借助另外两个文件：
  • [config_database]（我们已进行过研究），该文件负责配置 ORM 和 [sqlalchemy]；
  • [config_layers]，用于配置应用程序的各层；

[config] 文件内容如下：

def configure(config: dict) -> dict:
    # [config] 拥有密钥 [sgbd]，其值为：
    # [mysql] 用于管理数据库 MySQL
    # [pgres] 用于管理数据库 PostgreSQL

    import os

    # 步骤 1 ---
    # 设置应用程序的 Python 路径

    # 此脚本文件的绝对路径
    script_dir = os.path.dirname(os.path.abspath(__file__))

    # root_dir（可能需要更改）
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"

    # 依赖项的绝对路径
    absolute_dependencies = [
        # 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",
        # AdminData, ImpôtsError, TaxPayer
        f"{root_dir}/impots/v04/entities",
        # BaseEntity, MyException
        f"{root_dir}/classes/02/entities",
        # 本地文件夹
        f"{script_dir}",
        f"{script_dir}/../../interfaces",
        f"{script_dir}/../../services",
        f"{script_dir}/../../entities",
    ]

    # 设置系统路径
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    # 步骤 2 ------
    # 完成应用程序配置
    config.update({
        # 数据文件的绝对路径
        "admindataFilename": f"{script_dir}/../../data/input/admindata.json"
    })

    # 步骤 3 ------
    # 数据库配置
    import config_database
    config = config_database.configure(config)

    # 步骤 4 ------
    # 应用程序层的实例化
    import config_layers
    config = config_layers.configure(config)

    # 提交配置
    return config

• 第 8-36 行：构建应用程序的 Python 路径；
• 第 38-43 行：在配置中添加文件 [admindata.json] 的路径；
• 第 45-48 行：配置 [sqlalchemy]；
• 第 50-53 行：实例化应用程序的层；
• 第 56 行：返回全局配置；

文件 [config_layers] 内容如下：

def configure(config: dict) -> dict:
    # [dao] 层实例化
    from DaoTransferAdminDataFromJsonFile2Database import DaoTransferAdminDataFromJsonFile2Database
    config['dao'] = DaoTransferAdminDataFromJsonFile2Database(config)

    # 返回配置
    return config

• 第 3-4 行：实例化 [dao] 层。我们看到，[DaoTransferAdminDataFromJsonFile2Database] 类的构造函数期望将应用程序的通用配置字典作为参数；
• 第 4 行：将 [dao] 层的引用放入配置中；
• 第 7 行：提交配置；

20.1.7. 应用程序的脚本 [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 ImpôtsError import ImpôtsError

# 获取层 [dao]
dao = config["dao"]

# 代码
try:
    # 将数据导入数据库
    dao.transfer_admindata_in_database()
except ImpôtsError as ex1:
    # 显示错误
    print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
    # 显示错误
    print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
    # 结束
    print("Terminé...")

• 第 1-10 行：等待一个参数。检查该参数是否存在且正确；
• 第 12-14 行：通过传入所选的 SGBD 类型作为参数，配置应用程序（通用设置、SQLAlchemy、层）；
• 第19-20行：需要使用[dao]层。将其加载；
• 第25行：将数据传输至数据库。方法[transfer_admindata_in_database]所需的所有信息均可在第20行[dao]层的属性中找到。该方法将从该处获取这些信息；

在执行完MySQL后，该层包含以下元素（phpMyAdmin）：

在 [3] 列中，可以看到 MySQL 为主键 [id] 分配的值。编号从 1 开始。上图是多次执行脚本后获得的屏幕截图。

使用数据库 PostgreSQL 时，结果如下：

• 右键单击 [1]，然后单击 [2-3]；
• 在 [4] 中，确实包含各税档的数据；

对常量表 [tbconstantes] 重复上述操作：

20.2. 应用 2：批处理模式下的税额计算

20.2.1. 架构

第 4 版的税款计算应用程序采用以下架构：

[dao] 层实现了 [InterfaceImpôtsDao] 接口。我们构建了一个实现该接口的类：

• [ImpôtsDaoWithAdminDataInJsonFile]，该类会从文件 jSON 中读取税务数据。这是第 3 版；

我们将通过一个新类 [ImpotsDaoWithTaxAdminDataInDatabase] 来实现 [InterfaceImpôtsDao] 接口，该类将从数据库中获取税务部门的数据。 [dao]层与之前一样，将结果写入jSON文件，并从文本文件中获取纳税人数据。 我们知道，只要继续遵循 [InterfaceImpôtsDao] 接口，[métier] 层就无需修改。

新架构如下：

20.2.2. 应用程序配置

配置文件 [config_database] 与应用程序 1 中的保持一致。配置 [config] 包含以下新内容：

    # 步骤 2------
    # 完成应用程序配置
    config.update({
        # 数据文件的绝对路径
        "admindataFilename": f"{script_dir}/../../data/input/admindata.json",
        "taxpayersFilename": f"{script_dir}/../../data/input/taxpayersdata.txt",
        "errorsFilename": f"{script_dir}/../../data/output/errors.txt",
        "resultsFilename": f"{script_dir}/../../data/output/résultats.json"
    })

• 第 6-8 行：应用程序 2 所用文本文件的绝对路径；

图层配置 [config_layers] 的变更如下：

def configure(config: dict) -> dict:
    # DAO层实例化
    from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
    config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)

    # 实例化 [métier] 层
    from ImpôtsMétier import ImpôtsMétier
    config['métier'] = ImpôtsMétier()

    # 提交配置
    return config

• 第 3-4 行：[dao] 层现由类 [ImpotsDaoWithAdminDataInDatabase] 实现。该类为新类，但实现了与应用练习第 4 版相同的 [InterfaceDao] 接口；
• 第7-8行：[métier]层由类[ImpôtsMétier]实现。这是应用练习第4版中使用的类；

20.2.3. [dao] 层

接口 [InterfaceImpôtsDao] 的实现类 [ImpotsDaoWithAdminDataInDatabase] 将如下所示：

# 导入
from sqlalchemy.exc import DatabaseError, IntegrityError, InterfaceError

from AbstractImpôtsDao import AbstractImpôtsDao
from AdminData import AdminData
from Constantes import Constantes
from ImpôtsError import ImpôtsError
from Tranche import Tranche


class ImpotsDaoWithAdminDataInDatabase(AbstractImpôtsDao):
    # 构造函数
    def __init__(self, config: dict):
        # config["taxPayersFilename"]：纳税人文本文件的名称
        # config["taxPayersResultsFilename"]：结果文件的名称
        # config["errorsFilename"]：记录在taxPayersFilename中发现的错误
        # config["database"]：数据库配置

        # 初始化 Parent 类
        AbstractImpôtsDao.__init__(self, config)
        # 参数存储
        self.__config = config
        # 管理数据
        self.__admindata = None

    # 接口实现
    def get_admindata(self):
        # admindata 是否已被保存？
        if self.__admindata:
            return self.__admindata
        # 向 BD 发出请求
        session = None
        config = self.__config
        try:
            # 一个会话
            database_config = config["database"]
            session = database_config["session"]

            # 读取税率区间表
            tranches = session.query(Tranche).all()

            # 读取常量表（仅1行）
            constantes = session.query(Constantes).first()

            # 创建 admindata 实例
            admindata = AdminData()
            # 在其中创建限额表 coeffR、coeffN
            limites = admindata.limites = []
            coeffr = admindata.coeffr = []
            coeffn = admindata.coeffn = []
            for tranche in tranches:
                limites.append(float(tranche.limite))
                coeffr.append(float(tranche.coeffr))
                coeffn.append(float(tranche.coeffn))
            # 向其中添加常量
            admindata.fromdict(constantes.asdict())
            # 保存 admindata
            self.__admindata = admindata
            # 返回该值
            return self.__admindata
        except (IntegrityError, DatabaseError, InterfaceError) as erreur:
            # 以另一种形式重新抛出异常
            raise ImpôtsError(27, f"{erreur}")
        finally:
            # 关闭会话
            if session:
                session.close()

注释

• 第 11 行：类 [ImpotsDaoWithAdminDataInDatabase] 继承了第 4 版中介绍的类 [AbstractImpôtsDao]。已知后者实现了该版本中介绍的接口 [InterfaceDao]。 正是遵循了该接口，才使我们无需更改 [métier] 层；
• 第 13 行：类构造函数接收应用程序配置字典作为参数；
• 第20行：父类 [] 被初始化。它部分实现了接口 [InterfaceDao]：
  • [get_taxpayers_data] 读取包含纳税人数据的文件 [taxpayersdata.txt]；
  • [write_taxpayers_results] 将结果写入文件 jSON [résultats.json]；
  • [get_admindata] 未实现；
• 第 22 行：保存作为参数传递的配置；
• 第 27 行：实现接口 [InterfaceDao] 中的方法 [get_admindata]：
• 第 28-30 行：方法 [get_admindata] 从税务管理部门获取数据，将其存储在类型为 [AdminData] 的对象中，并将该对象存储在 [self.__admindata] 中。 如果多次调用方法 [get_admindata]，则不会多次查询数据库。仅在第一次调用时进行查询。后续调用中，将返回对象 [self.__admindata]；
• 第 36-37 行：检索在应用程序配置过程中由 [config_database] 创建的会话 [sqlalchemy]；
• 第40行：将税率区间存入列表中；
• 第 43 行：获取税额计算常量；
• 第 46 行：创建 [AdminData] 类的实例。需注意该类继承自 [BaseEntity]；
• 第48-54行：初始化[AdminData]实例中的[limites, coeffr, coeffn]数组；
• 第 55-56 行：使用税额计算的常量初始化 [AdminData] 的其他属性。此前已特意将 [AdminData] 和 [Constantes] 类的属性命名为相同名称，这简化了代码；
• 第 57-58 行：将 [AdminData] 实例存储在 [dao] 层中，以便在后续调用 [get_admindata] 方法时返回该实例；
• 第 60 行：返回调用方代码所请求的值；
• 第 61-63 行：处理可能出现的错误；
• 第 64-67 行：数据库仅执行一次查询。因此可以关闭 [sqlalchemy] 会话；

20.2.4. [dao] 层的测试

在该应用程序的第 4 版中，我们构建了一个用于测试 [métier] 层的测试类。更准确地说，它同时测试了 [métier] 和 [dao] 这两个层。 我们沿用该测试来验证 [dao] 层是否按预期运行。实际上，[métier] 层并未发生变化。

[TestDaoMétier] 测试内容如下：

import unittest


class TestDaoMétier(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})
        métier.calculate_tax(taxpayer, admindata)
        # 验证
        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})
        métier.calculate_tax(taxpayer, admindata)
        # 校验
        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__':
    # 需要指定 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})
    # 业务层
    métier = config['métier']
    try:
        # 数据管理
        admindata = config['dao'].get_admindata()
    except BaseException as ex:
        # 显示
        print((f"L'erreur suivante s'est produite : {ex}"))
        # 结束
        sys.exit()
    # 将脚本接收的参数传递出去
    sys.argv.pop()
    # 执行测试方法
    print("tests en cours...")
   unittest.main()

• 我们不再赘述第 |test couche [métier] version 4| 段落中描述的11项测试；
• 第37-66行：我们将像运行普通应用程序一样执行测试脚本，而非作为UnitTest测试。第66行将调用UnitTest框架。 在之前的测试中，我们使用 [setUp] 方法来配置每个测试的执行。由于 [setUp] 函数在每次测试前都会被调用，因此我们需要重复执行相同的配置 11 次。 在此，我们仅进行一次配置。该配置包括定义全局变量 [métier]（第 53 行）和 [admindata]（第 56 行），这些变量随后将被 [TestDaoMétier] 方法（例如第 12 行）所使用；
• 第39-47行：测试脚本等待一个名为[mysql / pgres]的参数，该参数用于指定使用MySQL还是PostgreSQL数据库；
• 第50-51行：配置测试；
• 第 53 行：从配置中获取 [métier] 层；
• 第56行：对[dao]层执行相同操作。随后获取[admindata]实例，该实例封装了计算税款所需的数据；
• 测试表明，第66行的[unittest.main()]方法并未忽略脚本接收的[mysql / pgres]参数，而是赋予其了不同的含义。第63行确保该方法不再包含任何参数；

我们创建两个执行配置：

若执行其中任一配置，将得到以下结果：

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/v05/tests/TestDaoMétier.py mysql
tests en cours...
...........
----------------------------------------------------------------------
Ran 11 tests in 0.001s

OK

Process finished with exit code 0

• 第5行和第7行：11个测试均通过；

需注意，这些测试仅验证了11种税额计算情况。但测试通过已足以让我们对[dao]层建立信心。

20.2.5. 主脚本

主脚本 [main] 与第 4 版中的相同：

# 等待 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})

# 已建立 syspath - 可以进行导入
from ImpôtsError import ImpôtsError

# 获取应用程序的层（它们已实例化）
dao = config["dao"]
métier = config["métier"]

try:
    # 获取税率区间
    admindata = dao.get_admindata()
    # 读取纳税人数据
    taxpayers = dao.get_taxpayers_data()["taxpayers"]
    # 纳税人？
    if not taxpayers:
        raise ImpôtsError(57, f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
    # 计算纳税人的税款
    for taxPayer in taxpayers:
        # taxPayer 既是输入参数也是输出参数
        # taxPayer 将被修改
        métier.calculate_tax(taxPayer, admindata)
    # 将结果写入文本文件
    dao.write_taxpayers_results(taxpayers)
except ImpôtsError as erreur:
    # 显示错误
    print(f"L'erreur suivante s'est produite : {erreur}")
finally:
    # 完成
    print("Travail terminé...")

注释

• 第1-10行：获取参数[mysql / pgres]，该参数指定了要使用的SGBD；
• 第12-14行：配置应用程序；
• 第16-17行：导入类[ImpôtsError]。第38行需要用到该类；
• 第19-21行：获取应用程序各层的引用；
• 第25行：向[dao]层请求税务管理数据。[métier]层需要这些数据来计算税款；
• 第27行：从列表中获取纳税人的数据（ID、婚姻状况、子女数、工资）；
• 第29-30行：若该列表为空，则抛出异常；
• 第32-35行：计算列表[taxpayers]中各条目的税额；
• 第37行：将结果写入文件jSON[résultats.json]；
• 第38-40行：处理可能出现的错误；

为了执行该脚本，我们创建了两个 |执行配置|：

文件 [résultats.json] 中获得的结果是第 4 版的。

20.3. 应用 3：交互式税额计算

现在我们介绍一个用于交互式计算税款的应用程序。这是对第4版应用程序2的移植。

• 脚本 [main] 通过 [ui] 层中的 [ui.run] 方法启动与用户的对话；
• [ui] 层：
  • 使用 [dao] 层获取用于计算税款的数据；
  • 要求用户提供待计算税款的纳税人相关信息；
  • 使用 [métier] 层进行此计算；

文件 [config_layers] 实例化了一个额外的层：

def configure(config: dict) -> dict:
    # 实例化 DAO 层
    from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
    config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)

    # 实例化 [métier] 层
    from ImpôtsMétier import ImpôtsMétier
    config['métier'] = ImpôtsMétier()

    # ui
    from ImpôtsConsole import ImpôtsConsole
    config['ui'] = ImpôtsConsole(config)

    # 返回配置
    return config

类 [ImpôtsConsole] 的第 11-12 行与 |版本 4| 中的内容相同。

主脚本 [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 ImpôtsError import ImpôtsError

# 获取层 [ui]
ui = config["ui"]

# 代码
try:
    # 执行 [ui] 层
    ui.run()
except ImpôtsError as ex1:
    # 显示错误信息
    print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
    # 显示错误信息
    print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
    # 在所有情况下均执行
    print("Travail terminé...")

• 第 1-10 行，脚本等待一个名为 [mysql / pgres] 的参数，该参数指定要使用的 SGBD；
• 第 12-14 行：配置应用程序；
• 第 19-20 行：从配置中获取 [ui] 层；
• 第25行：执行该图层；

结果与|第4版|完全一致。这理所当然，因为第5版完全遵循了第4版的所有接口规范。