20. Application exercise: version 5
We will develop three applications:
- Application 1 will initialize the database that will replace the file [admindata.json] from version 4;
- Application 2 will calculate taxes in batch mode;
- Application 3 will calculate taxes in interactive mode;
20.1. Application 1: Database Initialization
Application 1 will have the following architecture:
This is an evolution of the architecture of version 4 (section |Version 4|): tax data will be stored in a database instead of in a jSON file. The [dao] layer will be updated to implement this change.
20.1.1. The [admindata.json] file
The [admindata.json] file is the same as it was in version 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
}
We will use the keys from this dictionary as columns in the database.
20.1.2. Creating the Databases
As shown in the section |Creating a database MySQL|, we create a database named MySQL owned by user [admimpots] with password [mdpimpots]. In [phpMyAdmin], this results in the following:
Similarly, as shown in the section |Creating a database PostgreSQL|, we create a database named PostgreSQL owned by the user [admimpots] with password [mdpimpots]. In [pgAdmin], this results in the following:
The databases are created but currently have no tables. These will be created by ORM and [sqlalchemy].
20.1.3. Entities mapped by [sqlalchemy]
We will create two tables to encapsulate the data from [admindata.json]:
Defined by [sqlalchemy], the table [tbtranches] will collect data from the arrays [limites, coeffr, coeffn] in the dictionary [admindata.json]:
# 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)
)
Defined by [sqlalchemy], the table [tbconstantes] will contain the constants from the dictionary [admindata.json]:
# 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)
)
The entities that will be mapped to these two tables are as follows:
The [Constantes] entity encapsulates the constants from the [admindata.json] dictionary:
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"]
- line 5: the class [Constantes] extends the class [BaseEntity];
- line 7: via mapping [sqlalchemy], class [Constante] will receive property [_sa_instance_state]. We exclude it from the entity’s dictionary [asdict];
- lines 11–23: the entity’s properties. We have reused the names used in the [admindata.json] dictionary to simplify code writing;
The entity [Tranche] encapsulates a row from the three tables [limites, coeffr, coeffn] in the dictionary [admindata.json]:
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"]
- line 5: the class [Tranche] extends the class [BaseEntity];
- line 7: the property [_sa_instance_state] added by [sqlalchemy] is excluded from the entity's [asdict] dictionary;
- lines 10–12: the properties of the class;
The mapping between the entities [Constantes, Tranche] and the tables [constantes, tranches] will be as follows:
…
# 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 mappings are defined on lines 24–29. We have omitted the mappings between the properties of the mapped entities and the database tables. This is possible when the names of the table columns are the same as those of the properties they are to be associated with. For this reason, we have included the names of the mapped entities’ properties in the tables. This makes the code easier to write and understand;
20.1.4. The [sqlalchemy] configuration file
We have just detailed part of the configuration for [sqlalchemy]. The complete [config_database] file is as follows:
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
- line 1: the [configure] function receives a dictionary as a parameter, whose key [sgbd] tells it which SGBD to use: MySQL (mysql) or PostgreSQL (pgres);
- lines 6–12: the database required by the configuration is selected;
- lines 14–44: entity/table mappings. These mappings are simple because there is no relationship between the [tranches] and [constantes] tables. They are independent. Therefore, there are no foreign keys between them to manage;
- lines 46–51: the application’s [session] work session is created;
- lines 53–58: The relevant information is placed in the configuration dictionary, which is then returned;
20.1.5. The [dao] layer
Let’s return to the architecture of Application 1 to be built:
The [dao] [1] layer must read the [admindata.json] [2] file and transfer its contents to one of the [3, 4] databases;
The [dao] layer exposes the [1] interface and is implemented by the [2] class.
The [InterfaceDao4TransferAdminData2Database] interface is as follows:
# 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
- lines 8–10: the interface defines only one method, [transfer_admindata_in_database], with no parameters. Since this method requires parameters (which file?, which database?), this means that these parameters will be passed to the constructor of the classes implementing this interface;
The [DaoTransferAdminDataFromJsonFile2Database] class implements the [InterfaceDao4TransferAdminData2Database] interface as follows:
# 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:
# release session resources
if session:
session.close()
- line 13: the [DaoTransferAdminDataFromJsonFile2Database] class implements the [InterfaceDao4TransferAdminData2Database] interface;
- lines 15–17: the class constructor takes the configuration dictionary as a parameter. The following keys will be used:
- [admindataFilename] (line 27): the name of the jSON file containing the tax administration data to be transferred to the database;
- [database] line 32: the application’s [sqlalchemy] configuration;
- lines 34–37: deletion of tables [constantes] and [tranches] if they exist;
- lines 39-40: recreate the two tables;
- line 43: retrieve the [sqlalchemy] session present in the configuration;
- lines 45–51: the tables [limites, coeffr, coeffn] from the dictionary [admindata] are added to the session. To do this, instances of the entity [Tranche] are added to the session;
- lines 52-64: an instance of the [Constantes] entity is added to the session;
- lines 66-67: the session is validated. If the session data was not yet in the database, it is inserted at this point;
- lines 68–70: handling of any errors;
- lines 71-74: the session is closed. This is possible because the [dao] layer is used only once;
20.1.6. Application Configuration
The application is configured by three [1] files:
- [config] is the general configuration file. It configures the [main] application. It is assisted by the other two files:
- [config_database], which we have examined and which configures ORM and [sqlalchemy];
- [config_layers], which configures the application layers;
The [config] file is as follows:
def configure(config: dict) -> dict:
# [config] has the key [sgbd] which is worth:
# [mysql] to manage a MySQL database
# [pgres] to manage a PostgreSQL database
import os
# step 1 ---
# set up 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)
# we return the config
return config
- lines 8–36: build the application’s Python Path;
- lines 38-43: set the path to the [admindata.json] file in the configuration;
- lines 45-48: [sqlalchemy] configuration;
- lines 50–53: instantiate the application layers;
- line 56: return the general configuration;
The [config_layers] file is as follows:
def configure(config: dict) -> dict:
# instantiation layer [dao]
from DaoTransferAdminDataFromJsonFile2Database import DaoTransferAdminDataFromJsonFile2Database
config['dao'] = DaoTransferAdminDataFromJsonFile2Database(config)
# we return the config
return config
- lines 3-4: instantiation of the [dao] layer. We saw that the constructor of the [DaoTransferAdminDataFromJsonFile2Database] class expects the application’s general configuration dictionary as a parameter;
- line 4: the reference to the [dao] layer is added to the configuration;
- line 7: the configuration is returned;
20.1.7. The application’s [main] script
The main script [main] is as follows:
# 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 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é...")
- lines 1-10: we expect a parameter. We check that it is present and correct;
- lines 12-14: we configure the application (general, SQLAlchemy, layers) by passing the selected SGBD type as a parameter;
- lines 19-20: we will need the [dao] layer. We retrieve it;
- line 25: we perform the transfer to the database. All the information required by the [transfer_admindata_in_database] method is available in the properties of the [dao] layer from line 20. That is where it will retrieve it;
After execution with database MySQL, it contains the following elements (phpMyAdmin):
In column [3], we see the values assigned by MySQL to the primary key [id]. The numbering starts at 1. The screenshot above was obtained after running the script several times.
With the database PostgreSQL, the results are as follows:
- Right-click on [1], then on [2-3];
- in [4], the tax bracket data is present;
We repeat the same process for the constants table [tbconstantes]:
20.2. Application 2: Tax calculation in batch mode
20.2.1. Architecture
The tax calculation application for version 4 used the following architecture:
The [dao] layer implements the [InterfaceImpôtsDao] interface. We built a class that implements this interface:
- [ImpôtsDaoWithAdminDataInJsonFile], which retrieved tax data from a jSON file. This was version 3;
We will implement the [InterfaceImpôtsDao] interface using a new class, [ImpotsDaoWithTaxAdminDataInDatabase], which will retrieve data from the tax authority in a database. The [dao] layer, as before, will write the results to a jSON file and retrieve taxpayer data from a text file. We know that if we continue to adhere to the [InterfaceImpôtsDao] interface, the [métier] layer will not need to be modified.
The new architecture will be as follows:
20.2.2. Application Configuration
The [config_database] configuration file remains unchanged from the version in Application 1. The [config] configuration includes new elements:
# step 2 ------
# complete application configuration
config.update({
# absolute paths for data files
"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"
})
- lines 6–8: the absolute paths of the text files used by application 2;
The configuration of the [config_layers] layers changes as follows:
def configure(config: dict) -> dict:
# instantiation layer dao
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# instantiation layer [métier]
from ImpôtsMétier import ImpôtsMétier
config['métier'] = ImpôtsMétier()
# we return the config
return config
- lines 3-4: the [dao] layer is now implemented by the [ImpotsDaoWithAdminDataInDatabase] class. This class is new but implements the same [InterfaceDao] interface as version 4 from the application exercise;
- lines 7–8: The [métier] layer is implemented by the [ImpôtsMétier] class. This is the class used in version 4 from the application exercise;
20.2.3. The [dao] layer
The implementation class [ImpotsDaoWithAdminDataInDatabase] for the interface [InterfaceImpôtsDao] will be as follows:
# 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"]: records 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):
# has admindata been 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()
Notes
- line 11: the class [ImpotsDaoWithAdminDataInDatabase] inherits from the class [AbstractImpôtsDao] presented in version 4. We know that the latter implements the [InterfaceDao] interface presented in this same version. It is compliance with this interface that allows us to leave the [métier] layer unchanged;
- line 13: the class constructor receives the application configuration dictionary as a parameter;
- line 20: the parent class [] is initialized. It partially implements the [InterfaceDao] interface:
- [get_taxpayers_data] reads the file [taxpayersdata.txt], which contains taxpayer data;
- [write_taxpayers_results] writes the results to the file jSON [résultats.json];
- [get_admindata] is not implemented;
- line 22: the configuration passed as parameters is stored;
- line 27: implementation of the [get_admindata] method of the [InterfaceDao] interface:
- lines 28–30: the method [get_admindata] retrieves data from the tax administration into an object of type [AdminData] and stores this object in [self.__admindata]. If the [get_admindata] method is called multiple times, the database is not queried multiple times. It is queried only the first time. On subsequent calls, the [self.__admindata] object is returned;
- lines 36–37: the session [sqlalchemy], which was created during application configuration by [config_database], is retrieved;
- line 40: the tax brackets are retrieved into a list;
- lines 43: the tax calculation constants are retrieved;
- line 46: an instance of the [AdminData] class is created. Note that it derives from [BaseEntity];
- lines 48–54: The arrays [limites, coeffr, coeffn] of the instance [AdminData] are initialized;
- lines 55–56: the other properties of [AdminData] are initialized with the tax calculation constants. Care was taken to give the same names to the properties of the classes [AdminData] and [Constantes], which simplifies the code;
- lines 57–58: the [AdminData] instance is stored in the [dao] layer to be returned during subsequent calls to the [get_admindata] method;
- line 60: the value requested by the calling code is returned;
- lines 61–63: Handling any errors;
- lines 64–67: only a single query is made to the database. The [sqlalchemy] session can therefore be closed;
20.2.4. Testing the [dao] layer
In version 4 of this application, we had built a test class for the [métier] layer. More precisely, it tested both the [métier] and [dao] layers. We are reusing this test to verify that the [dao] layer functions as expected. In fact, the [métier] layer remains unchanged.
The [TestDaoMétier] test is as follows:
import unittest
class TestDaoMétier(unittest.TestCase):
def test_1(self) -> None:
from TaxPayer import TaxPayer
# { 'married': 'yes', 'children': 2, 'salary': 55555,
# tax': 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)
# check
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
# { 'married': 'yes', 'children': 3, 'salary': 200000,
# tax': 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)
# checks
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__':
# 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})
# business layer
métier = config['métier']
try:
# admindata
admindata = config['dao'].get_admindata()
except BaseException as ex:
# display
print((f"L'erreur suivante s'est produite : {ex}"))
# end
sys.exit()
# we take the parameter received by the script
sys.argv.pop()
# test methods are executed
print("tests en cours...")
unittest.main()
- We will not revisit the 11 tests described in section |test layer [métier] version 4|;
- lines 37–66: we will run the test script as a normal application and not as a test UnitTest. Line 66 is what will invoke the UnitTest framework. In the previous tests, we used the [setUp] method to configure the execution of each test. We repeated the same configuration 11 times since the [setUp] function is executed before each test. Here, we perform the configuration once. It consists of defining global variables [métier] on line 53 and [admindata] on line 56, which will then be used by the methods of [TestDaoMétier], for example on line 12;
- lines 39–47: the test script expects a parameter [mysql / pgres] that indicates whether to use database MySQL or PostgreSQL;
- lines 50–51: the test is configured;
- line 53: the [métier] layer is retrieved from the configuration;
- line 56: the same is done with the [dao] layer. The [admindata] instance, which encapsulates the data needed to calculate the tax, is then retrieved;
- tests showed that the [unittest.main()] method on line 66 did not ignore the [mysql / pgres] parameter received by the script but assigned it a different meaning. Line 63 ensures that this method no longer has any parameters;
We create two execution configurations:
If we run either of these two configurations, we get the following results:
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
- Lines 5 and 7: all 11 tests were successful;
Note that these tests only verify 11 cases of tax calculation. Their success may nevertheless be sufficient to give us confidence in the [dao] layer.
20.2.5. The main script
The main script [main] is the same as in version 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é...")
Notes
- lines 1-10: we retrieve the parameter [mysql / pgres], which specifies the SGBD to use;
- lines 12-14: the application is configured;
- lines 16-17: the [ImpôtsError] class is imported. We need it on line 38;
- lines 19–21: references to the application layers are retrieved;
- line 25: we request tax administration data from the [dao] layer. The [métier] layer needs this for tax calculation;
- line 27: we retrieve the taxpayer data (id, marital status, children, salary) into a list;
- lines 29–30: if this list is empty, an exception is thrown;
- lines 32–35: calculate the tax for the items in the [taxpayers] list;
- line 37: write the results to the file jSON[résultats.json];
- lines 38-40: handle any errors;
To run the script, we create two |execution configurations|:
The results obtained in the file [résultats.json] are those of version 4.
20.3. Application 3: Tax calculation in interactive mode
We now introduce the application that allows for interactive tax calculation. This is a port of Application 2 from version 4.
- The script [main] initiates user interaction using the method [ui.run] from the layer [ui];
- The [ui] layer:
- uses the [dao] layer to retrieve the data needed to calculate the tax;
- asks the user for information regarding the taxpayer for whom the tax is to be calculated;
- uses layer [métier] to perform this calculation;
The [config_layers] file instantiates an additional layer:
def configure(config: dict) -> dict:
# instantiation layer dao
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# instantiation layer [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)
# we return the config
return config
The [ImpôtsConsole] class, lines 11–12, is the same as in |version 4|.
The main script [main] is as follows:
# 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
# retrieve the [ui] layer
ui = config["ui"]
# code
try:
# execute [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é...")
- lines 1-10: the script expects a parameter [mysql / pgres] that specifies which SGBD to use;
- lines 12-14: the application is configured;
- lines 19-20: the [ui] layer is retrieved from the configuration;
- line 25: it is executed;
The results are identical to those of |version 4|. This is to be expected, since all the interfaces of version 4 were preserved in version 5.