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] 的 MySQL 数据库，所有者为 [admimpots]，密码为 [mdpimpots]。在 [phpMyAdmin] 中，其界面如下所示：
同样地，如“创建 PostgreSQL 数据库”一节中所示，我们创建了一个名为 [dbimpots-2019] 的 PostgreSQL 数据库，该数据库由用户 [admimpots] 拥有，密码为 [mdpimpots]。在 [pgAdmin] 中，其显示如下：
数据库已创建，但目前尚无表。这些表将由 [sqlalchemy] ORM 生成。
20.1.3. 由 [sqlalchemy] 映射的实体
我们将创建两个表来封装 [admindata.json] 中的数据：
由 [sqlalchemy] 定义的 [tbtranches] 表将收集 [admindata.json] 字典中 [limites、coeffr、coeffn] 数组的数据：
# la table des tranches de l'impôt
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] 字典的常量：
# la table des constantes
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)
)
将映射到这两个表的实体如下：
[Constants] 实体封装了 [admindata.json] 字典中的常量：
| from BaseEntity import BaseEntity
# tax administration data container class
class Constantes(BaseEntity):
# keys excluded from class state
excluded_keys = ["_sa_instance_state"]
# authorized keys
@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 行：[Constants] 类继承自 [BaseEntity] 类；
- 第 7 行：通过 [sqlalchemy] 映射，[Constante] 类将获得 [_sa_instance_state] 属性。我们将其从实体的 [asdict] 字典中排除；
- 第 11–23 行：实体的属性。我们复用了 [admindata.json] 字典中的名称，以便更轻松地编写代码；
[Tranche] 实体封装了 [admindata.json] 字典中 [limites, coeffr, coeffn] 这三个数组中的一行数据：
| from BaseEntity import BaseEntity
# tax administration data container class
class Tranche(BaseEntity):
# keys excluded from class state
excluded_keys = ["_sa_instance_state"]
# authorized keys
@staticmethod
def get_allowed_keys() -> list:
return ["id", "limite", "coeffr", "coeffn"]
|
- 第 5 行：[Tranche] 类继承自 [BaseEntity] 类；
- 第 7 行：由 [sqlalchemy] 添加的属性 [_sa_instance_state] 被排除在实体的 [asdict] 字典之外；
- 第 10–12 行：类属性；
[Constants, Slice] 实体与 [constants, slices] 表之间的映射关系如下：
| …
# the constants table
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)
)
# tax bracket table
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)
)
# mappings
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 configuration
from sqlalchemy import create_engine, Table, Column, Integer, MetaData, Float
from sqlalchemy.orm import mapper, sessionmaker
# connection chains to the databases used
connection_strings = {
'mysql': "mysql+mysqlconnector://admimpots:mdpimpots@localhost/dbimpots-2019",
'pgres': "postgresql+psycopg2://admimpots:mdpimpots@localhost/dbimpots-2019"
}
# connection chain to the database used
engine = create_engine(connection_strings[config['sgbd']])
# metadata
metadata = MetaData()
# the constants table
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)
)
# tax bracket table
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)
)
# mappings
from Tranche import Tranche
mapper(Tranche, tranches_table)
from Constantes import Constantes
mapper(Constantes, constantes_table)
# the factory session
session_factory = sessionmaker()
session_factory.configure(bind=engine)
# a session
session = session_factory()
# certain information is recorded
config['database'] = {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
"constantes_table": constantes_table, "session": session}
# result
return config
|
- 第 1 行：[configure] 函数接收一个字典作为参数，其 [dbms] 键指定要使用的数据库管理系统（DBMS）：MySQL（mysql）或 PostgreSQL（pgres）；
- 第 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] 接口如下：
| # imports
from abc import ABC, abstractmethod
# interface InterfaceImpôtsUI
class InterfaceDao4TransferAdminData2Database(ABC):
# transfer tax data to a database
@abstractmethod
def transfer_admindata_in_database(self:object):
pass
|
- 第 8–10 行：该接口仅定义了一个无参数的方法 [transfer_admindata_in_database]。由于该方法需要参数（哪个文件？哪个数据库？），这意味着这些参数将被传递给实现该接口的类的构造函数；
类 [DaoTransferAdminDataFromJsonFile2Database] 如下所示实现了接口 [InterfaceDao4TransferAdminData2Database]：
| # imports
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):
# manufacturer
def __init__(self, config: dict):
self.config = config
# transfer
def transfer_admindata_in_database(self) -> None:
# initializations
session = None
config = self.config
try:
# we retrieve data from the tax authorities
with codecs.open(config["admindataFilename"], "r", "utf8") as fd:
# transfer content to a dictionary
admindata = json.load(fd)
# retrieve the database configuration
database = config["database"]
# delete the two tables from the database
# checkfirst=True: first checks that the table exists
database["tranches_table"].drop(database["engine"], checkfirst=True)
database["constantes_table"].drop(database["engine"], checkfirst=True)
# recreate tables from mappings
database["metadata"].create_all(database["engine"])
# the current [sqlalchemy] session
session = database["session"]
# fill in the tax bracket table
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]}))
# fill in the constants table
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"]
}))
# session validation [sqlalchemy]
session.commit()
except (IntegrityError, DatabaseError, InterfaceError) as erreur:
# we relaunch the exception in another form
raise ImpôtsError(17, f"{erreur}")
finally:
# session resources are released
if session:
session.close()
|
- 第 13 行：类 [DaoTransferAdminDataFromJsonFile2Database] 实现了接口 [InterfaceDao4TransferAdminData2Database]；
- 第 15–17 行：类构造函数将配置字典作为参数。将使用以下键：
- [admindataFilename]（第 27 行）：包含待传输至数据库的税务管理数据的 JSON 文件名；
- [database] 第 32 行：应用程序的 [sqlalchemy] 配置；
- 第 34–37 行：若 [constants] 和 [brackets] 表存在，则将其删除；
- 第 39–40 行：重新创建这两个表；
- 第 43 行：从配置中获取 [sqlalchemy] 会话；
- 第 45–51 行：将 [admindata] 字典中的数组 [limits, coeffr, coeffn] 添加到会话中。为此，将 [Tranche] 实体的实例添加到会话中；
- 第 52–64 行：将 [Constantes] 实体的实例添加到会话中；
- 第 66–67 行：对会话进行验证。如果会话数据尚未存在于数据库中，则在此处插入；
- 第 68–70 行：错误处理；
- 第 71–74 行：关闭会话。这是因为 [dao] 层仅被调用一次；
20.1.6. 应用程序配置
该应用程序通过三个文件进行配置 [1]：
- [config] 是通用配置文件。它用于配置 [main] 应用程序。该文件由另外两个文件辅助：
- [config_database]，我们已对其进行过探讨，该文件用于配置 ORM [sqlalchemy]；
- [config_layers]，用于配置应用程序层；
[config] 文件内容如下：
| def configure(config: dict) -> dict:
# [config] has the key [sgbd]:
# [mysql] to manage a MySQL database
# [pgres] to manage a PostgreSQL database
import os
# step 1 ---
# establish the application's Python Path
# absolute path of this script's folder
script_dir = os.path.dirname(os.path.abspath(__file__))
# root_dir (change if necessary)
root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
# absolute paths of dependencies
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",
# local files
f"{script_dir}",
f"{script_dir}/../../interfaces",
f"{script_dir}/../../services",
f"{script_dir}/../../entities",
]
# set the syspath
from myutils import set_syspath
set_syspath(absolute_dependencies)
# step 2 ------
# complete application configuration
config.update({
# absolute paths for data files
"admindataFilename": f"{script_dir}/../../data/input/admindata.json"
})
# step 3 ------
# database configuration
import config_database
config = config_database.configure(config)
# step 4 ------
# instantiation of application layers
import config_layers
config = config_layers.configure(config)
# return the config
return config
|
- 第 8–36 行：构建应用程序的 Python 路径；
- 第 38–43 行：将 [admindata.json] 文件的路径添加到配置中；
- 第 45–48 行：[SQLAlchemy] 配置；
- 第 50–53 行：实例化应用层；
- 第 56 行：返回通用配置；
[config_layers] 文件内容如下：
| def configure(config: dict) -> dict:
# layer instantiation [dao]
from DaoTransferAdminDataFromJsonFile2Database import DaoTransferAdminDataFromJsonFile2Database
config['dao'] = DaoTransferAdminDataFromJsonFile2Database(config)
# return the config
return config
|
- 第 3-4 行：实例化 [dao] 层。我们看到 [DaoTransferAdminDataFromJsonFile2Database] 类的构造函数期望将应用程序的通用配置字典作为参数；
- 第 4 行：将 [dao] 层的引用添加到配置中；
- 第 7 行：返回配置；
20.1.7. 应用程序的 [main] 脚本
主 [main] 脚本如下：
| # a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
sgbd = sys.argv[1].lower()
erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
print(f"syntaxe : {syntaxe}")
sys.exit()
# configure the application
import config
config = config.configure({'sgbd': sgbd})
# the syspath is set up - imports can be made
from ImpôtsError import ImpôtsError
# we recover the [dao] layer
dao = config["dao"]
# code
try:
# data transfer to the database
dao.transfer_admindata_in_database()
except ImpôtsError as ex1:
# error is displayed
print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
# error is displayed
print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
# end
print("Terminé...")
|
- 第 1–10 行：我们等待一个参数。我们检查该参数是否存在且正确；
- 第 12–14 行：通过将选定的 DBMS 类型作为参数传递，配置应用程序（通用设置、SQLAlchemy、层）；
- 第 19–20 行：我们需要 [dao] 层。我们将其检索出来；
- 第 25 行：我们执行向数据库的传输。第 20 行获取的 [dao] 层属性中包含 [transfer_admindata_in_database] 方法所需的所有信息。该方法将从该处获取这些信息；
在 MySQL 数据库上运行脚本后，其中包含以下元素（phpMyAdmin）：
第 [3] 列显示了 MySQL 为主键 [id] 分配的值。编号从 1 开始。上图是在多次运行脚本后截取的。
在 PostgreSQL 数据库中，结果如下：
- 右键单击[1]，然后单击[2-3]；
- 在[4]中，税率区间数据清晰显示；
对于常量表 [tbconstantes]，我们也进行同样的操作：
20.2. 应用 2：批量计算税款
20.2.1. 架构
第 4 版的税费计算应用程序采用了以下架构：
[dao] 层实现了 [InterfaceImpôtsDao] 接口。我们构建了一个实现该接口的类：
- [TaxDaoWithAdminDataInJsonFile]，该类从 JSON 文件中获取税务数据。这是第 3 版；
我们将使用新类 [ImpotsDaoWithTaxAdminDataInDatabase] 来实现 [InterfaceImpôtsDao] 接口，该类将从数据库中检索税务管理数据。与之前一样，[dao] 层将把结果写入 JSON 文件，并从文本文件中检索纳税人数据。 我们知道，只要继续遵循 [InterfaceImpôtsDao] 接口，[business] 层就无需修改。
新架构如下：
20.2.2. 应用程序配置
[config_database] 配置文件与应用程序 1 中的保持一致。[config] 配置中包含新的元素：
# étape 2 ------
# on complète la configuration de l'application
config.update({
# chemins absolus des fichiers de données
"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 layer instantiation
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# instantiation layer [business]
from ImpôtsMétier import ImpôtsMétier
config['métier'] = ImpôtsMétier()
# return the config
return config
|
- 第 3-4 行：[dao] 层现由 [TaxDaoWithAdminDataInDatabase] 类实现。该类是新添加的，但实现了与应用练习第 4 版相同的 [DaoInterface] 接口；
- 第 7-8 行：[business] 层由 [ImpôtsMétier] 类实现。这是应用练习第 4 版中使用的类；
20.2.3. [DAO] 层
接口 [InterfaceImpôtsDao] 的实现类 [ImpotsDaoWithAdminDataInDatabase] 将如下所示：
| # imports
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):
# manufacturer
def __init__(self, config: dict):
# config["taxPayersFilename"]: name of the taxpayer text file
# config["taxPayersResultsFilename"]: name of the jSON results file
# config["errorsFilename"]: saves errors found in taxPayersFilename
# config["database"]: database configuration
# parent class initialization
AbstractImpôtsDao.__init__(self, config)
# parameter memory
self.__config = config
# admindata
self.__admindata = None
# interface implementation
def get_admindata(self):
# admindata memorized?
if self.__admindata:
return self.__admindata
# make a query in BD
session = None
config = self.__config
try:
# a session
database_config = config["database"]
session = database_config["session"]
# read the table of tax brackets
tranches = session.query(Tranche).all()
# read the constants table (1 line only)
constantes = session.query(Constantes).first()
# create the admindata instance
admindata = AdminData()
# we create limtes arrays, 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))
# we add the constants
admindata.fromdict(constantes.asdict())
# admindata is memorized
self.__admindata = admindata
# we return the value
return self.__admindata
except (IntegrityError, DatabaseError, InterfaceError) as erreur:
# we relaunch the exception in another form
raise ImpôtsError(27, f"{erreur}")
finally:
# close session
if session:
session.close()
|
注释
- 第 11 行：[ImpotsDaoWithAdminDataInDatabase] 类继承自第 4 版中介绍的 [AbstractImpôtsDao] 类。我们知道，后者实现了该版本中介绍的 [InterfaceDao] 接口。正是对该接口的遵循，使我们能够保持 [business] 层不变；
- 第 13 行：类构造函数将应用程序配置字典作为参数接收；
- 第 20 行：父类 [] 被初始化。它部分实现了 [InterfaceDao] 接口：
- [get_taxpayers_data] 读取包含纳税人数据的 [taxpayersdata.txt] 文件；
- [write_taxpayers_results] 将结果写入 JSON 文件 [results.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 版中，我们为 [business] 层创建了一个测试类。更具体地说，它同时测试了 [business] 和 [DAO] 层。我们将复用此测试来验证 [DAO] 层是否按预期运行。不过，[business] 层保持不变。
[TestDaoMétier] 测试代码如下：
import unittest
class TestDaoMétier(unittest.TestCase):
def test_1(self) -> None:
from TaxPayer import TaxPayer
# {'marié': 'oui', 'enfants': 2, 'salaire': 55555,
# 'impôt': 2814, 'surcôte': 0, 'décôte': 0, 'réduction': 0, 'taux': 0.14}
taxpayer = TaxPayer().fromdict({"marié": "oui", "enfants": 2, "salaire": 55555})
métier.calculate_tax(taxpayer, admindata)
# vérification
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
# {'marié': 'oui', 'enfants': 3, 'salaire': 200000,
# 'impôt': 42842, 'surcôte': 17283, 'décôte': 0, 'réduction': 0, 'taux': 0.41}
taxpayer = TaxPayer().fromdict({'marié': 'oui', 'enfants': 3, 'salaire': 200000})
métier.calculate_tax(taxpayer, admindata)
# vérifications
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__':
# on attend un paramètre mysql ou 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()
# on configure l'application
import config
config = config.configure({'sgbd': sgbd})
# couche métier
métier = config['métier']
try:
# admindata
admindata = config['dao'].get_admindata()
except BaseException as ex:
# affichage
print((f"L'erreur suivante s'est produite : {ex}"))
# fin
sys.exit()
# on enève le paramètre reçu par le script
sys.argv.pop()
# on exécute les méthodes de test
print("tests en cours...")
unittest.main()
- 我们不再回顾 |[业务] 层测试第 4 版| 章节中描述的 11 个测试；
- 第37–66行：我们将把测试脚本作为普通应用程序运行，而非作为UnitTest。第66行将触发UnitTest框架。在之前的测试中，我们使用[setUp]方法来配置每个测试的执行。 由于 [setUp] 函数在每次测试前都会被执行，因此我们重复了相同的配置 11 次。在此，我们仅执行一次配置。该配置包括在第 53 行定义全局变量 [business] 以及在第 56 行定义 [admindata]，这些变量随后将被 [TestDaoBusiness] 的方法所使用，例如第 12 行；
- 第 39–47 行：测试脚本期望有一个 [mysql / pgres] 参数，用于指示正在使用 MySQL 还是 PostgreSQL 数据库；
- 第 50–51 行：配置测试；
- 第 53 行：从配置中获取 [business] 层；
- 第 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 版相同：
| # a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
sgbd = sys.argv[1].lower()
erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
print(f"syntaxe : {syntaxe}")
sys.exit()
# configure the application
import config
config = config.configure({'sgbd': sgbd})
# the syspath is set up - imports can be made
from ImpôtsError import ImpôtsError
# retrieve application layers (already instantiated)
dao = config["dao"]
métier = config["métier"]
try:
# tax bracket recovery
admindata = dao.get_admindata()
# reading taxpayer data
taxpayers = dao.get_taxpayers_data()["taxpayers"]
# taxpayers?
if not taxpayers:
raise ImpôtsError(57, f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
# tax calculation
for taxPayer in taxpayers:
# taxPayer is both an input and output parameteri
# taxPayer will be modified
métier.calculate_tax(taxPayer, admindata)
# writing results to a text file
dao.write_taxpayers_results(taxpayers)
except ImpôtsError as erreur:
# error display
print(f"L'erreur suivante s'est produite : {erreur}")
finally:
# completed
print("Travail terminé...")
|
注释
- 第 1-10 行：我们获取 [mysql / pgres] 参数，该参数指定要使用的数据库管理系统；
- 第 12–14 行：配置应用程序；
- 第 16–17 行：导入 [ImpôtsError] 类。我们在第 38 行需要用到它；
- 第 19–21 行：获取应用程序层的引用；
- 第25行：我们从[dao]层请求税务管理数据。[business]层需要这些数据来计算税款；
- 第 27 行：我们将纳税人的数据（id、已婚、子女、工资）检索到一个列表中；
- 第29–30行：如果该列表为空，则抛出异常；
- 第 32–35 行：计算 [taxpayers] 列表中各项的税款；
- 第 37 行：将结果写入 JSON 文件 [results.json]；
- 第 38–40 行：处理任何错误；
要运行该脚本，我们需要创建两个 |执行配置|：
[results.json] 文件中获得的结果来自第 4 版。
20.3. 应用 3：交互式模式下的税费计算
现在我们介绍一个支持交互式税费计算的应用程序。这是将版本 4 中的应用程序 2 移植而来的。
- [main]脚本通过[ui]层的[ui.run]方法启动用户对话；
- [ui] 层：
- 使用 [dao] 层检索计算税款所需的数据；
- 向用户询问待计算税款的纳税人相关信息；
- 利用 [business] 层执行税费计算；
[config_layers] 文件实例化了一个额外的层：
| def configure(config: dict) -> dict:
# dao layer instantiation
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# instantiation layer [business]
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 the config
return config
|
第 11–12 行中的 [ImpôtsConsole] 类与 |版本 4| 中的相同。
主脚本 [main] 如下所示：
| # a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
sgbd = sys.argv[1].lower()
erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
print(f"syntaxe : {syntaxe}")
sys.exit()
# configure the application
import config
config = config.configure({'sgbd': sgbd})
# syspath is configured - imports can be made
from ImpôtsError import ImpôtsError
# we recover the [ui] layer
ui = config["ui"]
# code
try:
# execution of the [ui] layer
ui.run()
except ImpôtsError as ex1:
# the error message is displayed
print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
# the error message is displayed
print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
# executed in all cases
print("Travail terminé...")
|
- 第 1-10 行：脚本期望接收一个 [mysql / pgres] 参数，用于指定要使用的数据库管理系统；
- 第 12-14 行：配置应用程序；
- 第 19-20 行：从配置中获取 [ui] 层；
- 第 25 行：执行该配置；
结果与 |版本 4| 完全一致。这本是理所当然的，因为版本 5 完整保留了版本 4 的所有接口。