20. Application exercise: version 5
We will develop three applications:
- Application 1 will initialize the database that will replace the [admindata.json] file 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 Version 4 architecture (see the |Version 4| section): tax data will be stored in a database instead of 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],
"half-share_income_limit": 1551,
"income_limit_single_for_reduction": 21037,
"couple_income_limit_for_reduction": 42074,
"half-share_reduction_value": 3797,
"single_discount_limit": 1196,
"couple_discount_ceiling": 1970,
"couple_tax_ceiling_for_discount": 2627,
"single_tax_ceiling_for_discount": 1595,
"maximum_10_percent_deduction": 12502,
"minimum_10_percent_deduction": 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 MySQL database|, we create a MySQL database named [dbimpots-2019] owned by the user [admimpots] with password [mdpimpots]. In [phpMyAdmin], this looks like the following:
Similarly, as shown in the section |Creating a PostgreSQL Database|, we create a PostgreSQL database named [dbimpots-2019] owned by the user [admimpots] with the password [mdpimpots]. In [pgAdmin], this looks like the following:
The databases have been created, but for now they have no tables. These will be built by the [sqlalchemy] ORM.
20.1.3. Entities mapped by [sqlalchemy]
We will create two tables to encapsulate the data from [admindata.json]:
Defined by [sqlalchemy], the [tbtranches] table will collect data from the [limites, coeffr, coeffn] arrays in the [admindata.json] dictionary:
# the tax bracket table
tranches_table = Table("tbtranches", metadata,
Column('id', Integer, primary_key=True),
Column('limit', Float, nullable=False),
Column('coeffr', Float, nullable=False),
Column('coeffn', Float, nullable=False)
)
Defined by [sqlalchemy], the [tbconstantes] table will contain the constants from the [admindata.json] dictionary:
# the constants table
constants_table = Table("tbconstants", metadata,
Column('id', Integer, primary_key=True),
Column('half-share_income_limit', Float, nullable=False),
Column('single_income_limit_for_reduction', Float, nullable=False),
Column('couple_income_limit_for_reduction', Float, nullable=False),
Column('half-share_reduction_value', Float, nullable=False),
Column('single_discount_limit', Float, nullable=False),
Column('couple_discount_limit', Float, nullable=False),
Column('single_tax_threshold_for_discount', Float, nullable=False),
Column('couple_tax_ceiling_for_discount', Float, nullable=False),
Column('maximum_10_percent_deduction', Float, nullable=False),
Column('minimum_10_percent_deduction', Float, nullable=False)
)
The entities that will be mapped to these two tables are as follows:
The [Constants] entity encapsulates the constants from the [admindata.json] dictionary:
from BaseEntity import BaseEntity
# container class for tax administration data
class Constants(BaseEntity):
# keys excluded from the class state
excluded_keys = ["_sa_instance_state"]
# allowed keys
@staticmethod
def get_allowed_keys() -> list:
return ["id",
"half-share_qf_threshold",
"single_income_limit_for_reduction",
"couple_income_limit_for_reduction",
"half-share_reduction_value",
"single_discount_limit",
"couple_discount_limit",
"couple_discount_limit",
"single_tax_threshold_for_discount",
"couple_tax_ceiling_for_discount",
"maximum_10_percent_deduction",
"10-percent-minimum-discount"]
- line 5: the [Constants] class extends the [BaseEntity] class;
- line 7: via [sqlalchemy] mapping, the [Constante] class will receive the [_sa_instance_state] property. We exclude it from the entity’s [asdict] dictionary;
- lines 11–23: the entity’s properties. We have reused the names used in the [admindata.json] dictionary to make writing the code easier;
The [Tranche] entity encapsulates a row from the three arrays [limites, coeffr, coeffn] in the [admindata.json] dictionary:
from BaseEntity import BaseEntity
# container class for tax administration data
class Tranche(BaseEntity):
# keys excluded from the class state
excluded_keys = ["_sa_instance_state"]
# allowed keys
@staticmethod
def get_allowed_keys() -> list:
return ["id", "limit", "coeffr", "coeffn"]
- line 5: the [Tranche] class extends the [BaseEntity] class;
- line 7: the property [_sa_instance_state] added by [sqlalchemy] is excluded from the entity's [asdict] dictionary;
- lines 10–12: the class properties;
The mapping between the [Constants, Slice] entities and the [constants, slices] tables will be as follows:
…
# the constants table
constants_table = Table("tbconstants", metadata,
Column('id', Integer, primary_key=True),
Column('half-share_income_limit', Float, nullable=False),
Column('single_income_limit_for_reduction', Float, nullable=False),
Column('couple_income_limit_for_reduction', Float, nullable=False),
Column('half-share_reduction_value', Float, nullable=False),
Column('single_discount_limit', Float, nullable=False),
Column('couple_tax_deduction_limit', Float, nullable=False),
Column('tax_cap_for_single_tax_deduction', Float, nullable=False),
Column('couple_tax_threshold_for_discount', Float, nullable=False),
Column('maximum_10_percent_deduction', Float, nullable=False),
Column('minimum_10_percent_deduction', Float, nullable=False)
)
# the tax bracket table
bracket_table = Table("tbbrackets", metadata,
Column('id', Integer, primary_key=True),
Column('limit', Float, nullable=False),
Column('coeffr', Float, nullable=False),
Column('coeffn', Float, nullable=False)
)
# mappings
from Tranche import Tranche
mapper(Tranche, tranches_table)
from Constants import Constants
mapper(Constants, constants_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 [sqlalchemy] configuration. 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 strings for the databases in use
connection_strings = {
'mysql': "mysql+mysqlconnector://admimpots:mdpimpots@localhost/dbimpots-2019",
'pgres': "postgresql+psycopg2://admimpots:mdpimpots@localhost/dbimpots-2019"
}
# connection string for the database in use
engine = create_engine(connection_strings[config['sgbd']])
# metadata
metadata = MetaData()
# the constants table
constants_table = Table("tbconstants", metadata,
Column('id', Integer, primary_key=True),
Column('half-share_income_limit', Float, nullable=False),
Column('single_income_limit_for_reduction', Float, nullable=False),
Column('couple_income_limit_for_reduction', Float, nullable=False),
Column('half-share_reduction_value', Float, nullable=False),
Column('single_discount_limit', Float, nullable=False),
Column('couple_discount_limit', Float, nullable=False),
Column('single_tax_threshold_for_discount', Float, nullable=False),
Column('couple_tax_ceiling_for_discount', Float, nullable=False),
Column('maximum_10_percent_deduction', Float, nullable=False),
Column('minimum_10_percent_deduction', Float, nullable=False)
)
# the tax bracket table
bracket_table = Table("tbbrackets", metadata,
Column('id', Integer, primary_key=True),
Column('limit', Float, nullable=False),
Column('coeffr', Float, nullable=False),
Column('coeffn', Float, nullable=False)
)
# mappings
from Tranche import Tranche
mapper(Tranche, tranches_table)
from Constants import Constants
mapper(Constants, constants_table)
# the session factory
session_factory = sessionmaker()
session_factory.configure(bind=engine)
# a session
session = session_factory()
# We store some information
config['database'] = {"engine": engine, "metadata": metadata, "table_slices": table_slices,
"table_constants": table_constants, "session": session}
# result
return config
- line 1: the [configure] function receives a dictionary as a parameter, whose [dbms] key tells it which DBMS to use: MySQL (mysql) or PostgreSQL (pgres);
- lines 6–12: the database specified 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: create the application’s working session [session];
- 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] layer [1] must read the [admindata.json] file [2] and transfer its contents to one of the databases [3, 4];
The [dao] layer provides the interface [1] and is implemented by the class [2].
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 class [DaoTransferAdminDataFromJsonFile2Database] implements the interface [InterfaceDao4TransferAdminData2Database] as follows:
# imports
import codecs
import json
from sqlalchemy.exc import DatabaseError, IntegrityError, InterfaceError
from Constants import Constants
from TaxError import TaxError
from InterfaceDao4TransferAdminData2Database import InterfaceDao4TransferAdminData2Database
from Slice import Slice
class DaoTransferAdminDataFromJsonFile2Database(InterfaceDao4TransferAdminData2Database):
# constructor
def __init__(self, config: dict):
self.config = config
# transfer
def transfer_admindata_in_database(self) -> None:
# initializations
session = None
config = self.config
try:
# retrieve data from the tax administration
with codecs.open(config["admindataFilename"], "r", "utf8") as fd:
# transfer the 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 if the table exists
database["tranches_table"].drop(database["engine"], checkfirst=True)
database["constants_table"].drop(database["engine"], checkfirst=True)
# recreate the tables from the mappings
database["metadata"].create_all(database["engine"])
# the current [SQLAlchemy] session
session = database["session"]
# populate the tax bracket table
limits = admindata["limits"]
coeffr = admindata["coeffr"]
coeffn = admindata["coeffn"]
for i in range(len(limits)):
session.add(Tranche().fromdict(
{"limit": limits[i], "coeffr": coeffr[i], "coeffn": coeffn[i]}))
# populate the constants table
session.add(Constants().fromdict({
'half-share_tax_threshold': admindata["half-share_tax_threshold"],
'single_income_threshold_for_reduction': admindata["single_income_threshold_for_reduction"],
'couple_income_limit_for_reduction': admindata["couple_income_limit_for_reduction"],
'half-share_reduction_value': admindata["half-share_reduction_value"],
'single_discount_limit': admindata["single_discount_limit"],
'couple_discount_limit': admindata["couple_discount_limit"],
'single_tax_threshold_for_discount': admindata["single_tax_threshold_for_discount"],
'couple_tax_threshold_for_discount': admindata["couple_tax_threshold_for_discount"],
'maximum_10_percent_deduction': admindata["maximum_10_percent_deduction"],
'minimum_10_percent_deduction': admindata["minimum_10_percent_deduction"]
}))
# commit the session [sqlalchemy]
session.commit()
except (IntegrityError, DatabaseError, InterfaceError) as error:
# Re-raise the exception in a different form
raise TaxError(17, f"{error}")
finally:
# release session resources
if session:
session.close()
- line 13: the class [DaoTransferAdminDataFromJsonFile2Database] implements the interface [InterfaceDao4TransferAdminData2Database];
- 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 the [constants] and [brackets] tables if they exist;
- lines 39–40: recreate the two tables;
- line 43: retrieve the [sqlalchemy] session from the configuration;
- lines 45–51: the arrays [limits, coeffr, coeffn] from the [admindata] dictionary are added to the session. To do this, instances of the [Tranche] entity 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: error handling;
- 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 files [1]:
- [config] is the general configuration file. It configures the [main] application. It is assisted by the other two files:
- [config_database], which we have already examined and which configures the ORM [sqlalchemy];
- [config_layers], which configures the application layers;
The [config] file is as follows:
def configure(config: dict) -> dict:
# [config] has the key [db] with the value:
# [mysql] to manage a MySQL database
# [pgres] to manage a PostgreSQL database
import os
# Step 1 ---
# Set the application's Python path
# absolute path to this script's directory
script_dir = os.path.dirname(os.path.abspath(__file__))
# root_dir (may need to be changed)
root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
# absolute paths of dependencies
absolute_dependencies = [
# TaxDaoInterface, TaxBusinessInterface, TaxUIInterface
f"{root_dir}/taxes/v04/interfaces",
# TaxDaoAbstract, TaxConsole, TaxBusiness
f"{root_dir}/taxes/v04/services",
# AdminData, TaxError, TaxPayer
f"{root_dir}/taxes/v04/entities",
# BaseEntity, MyException
f"{root_dir}/classes/02/entities",
# local directories
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 the application configuration
config.update({
# absolute paths to data files
"admindataFilename": f"{script_dir}/../../data/input/admindata.json"
})
# Step 3 ------
# database configuration
import config_database
config = config_database.configure(config)
# Step 4 ------
# instantiating the application layers
import config_layers
config = config_layers.configure(config)
# return the configuration
return config
- lines 8–36: Build the application’s Python Path;
- lines 38–43: add the path to the [admindata.json] file to 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:
# instantiate the [dao] layer
from DaoTransferAdminDataFromJsonFile2Database import DaoTransferAdminDataFromJsonFile2Database
config['dao'] = DaoTransferAdminDataFromJsonFile2Database(config)
# 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: return the configuration;
20.1.7. The application’s [main] script
The main [main] script is as follows:
# Expects a mysql or pgres parameter
import sys
syntax = f"{sys.argv[0]} mysql / pgres"
error = len(sys.argv) != 2
if not error:
dbms = sys.argv[1].lower()
error = dbsystem != "mysql" and dbsystem != "pgres"
if error:
print(f"syntax: {syntax}")
sys.exit()
# configure the application
import config
config = config.configure({'db': db})
# The syspath is set—we can now import modules
from ImpôtsError import ImpôtsError
# Retrieve the [dao] layer
dao = config["dao"]
# code
try:
# transfer data to the database
dao.transfer_admindata_in_database()
except ImpôtsError as ex1:
# display the error
print(f"The following error occurred: {ex1}")
except BaseException as ex2:
# display the error
print(f"The following error 2 occurred: {ex2}")
finally:
# end
print("Done...")
- lines 1–10: We wait for a parameter. We check that it is present and correct;
- lines 12–14: We configure the application (general, SQLAlchemy, layers) by passing the chosen DBMS 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 running the script with the MySQL database, it contains the following elements (phpMyAdmin):
Column [3] shows the values assigned by MySQL to the primary key [id]. Numbering starts at 1. The screenshot above was taken after running the script several times.
With the PostgreSQL database, the results are as follows:
- Right-click on [1], then on [2-3];
- In [4], the tax bracket data is clearly displayed;
We do the same for the constants table [tbconstantes]:
20.2. Application 2: Calculating tax in batch mode
20.2.1. Architecture
The tax calculation application in version 4 used the following architecture:
The [dao] layer implements an interface [InterfaceImpôtsDao]. We built a class implementing this interface:
- [TaxDaoWithAdminDataInJsonFile], which retrieved tax data from a JSON file. That was version 3;
We will implement the [InterfaceImpôtsDao] interface using a new class [ImpotsDaoWithTaxAdminDataInDatabase] that will retrieve tax administration data from 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 [business] 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 the same as it was in Application 1. The [config] configuration includes new elements:
# step 2 ------
# complete the application configuration
config.update({
# absolute paths to 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/results.json"
})
- lines 6–8: the absolute paths of the text files used by application 2;
The configuration of the layers [config_layers] evolves as follows:
def configure(config: dict) -> dict:
# instantiate DAO layer
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# instantiate [business] layer
from BusinessTaxes import BusinessTaxes
config['business'] = BusinessTaxes()
# return the configuration
return config
- lines 3-4: the [dao] layer is now implemented by the [TaxDaoWithAdminDataInDatabase] class. This class is new but implements the same [DaoInterface] interface as version 4 of the application exercise;
- lines 7-8: the [business] layer is implemented by the [ImpôtsMétier] class. This is the class used in version 4 of 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 AbstractTaxDao import AbstractTaxDao
from AdminData import AdminData
from Constants import Constants
from TaxError import TaxError
from TaxBracket import TaxBracket
class TaxDaoWithAdminDataInDatabase(AbstractTaxDao):
# constructor
def __init__(self, config: dict):
# config["taxPayersFilename"]: the name of the taxpayer text file
# config["taxPayersResultsFilename"]: the name of the JSON file containing the results
# config["errorsFilename"]: saves any errors found in taxPayersFilename
# config["database"]: database configuration
# Initialization of the Parent class
AbstractTaxDao.__init__(self, config)
# storing parameters
self.__config = config
# admindata
self.__admindata = None
# interface implementation
def get_admindata(self):
# Has admindata been saved?
if self.__admindata:
return self.__admindata
# make a database query
session = None
config = self.__config
try:
# a session
database_config = config["database"]
session = database_config["session"]
# read the tax bracket table
brackets = session.query(Bracket).all()
# read the constants table (1 row)
constants = session.query(Constants).first()
# create the admindata instance
admindata = AdminData()
# create the limits, coeffR, and coeffN arrays
limits = admindata.limits = []
coeffr = admindata.coeffr = []
coeffn = admindata.coeffn = []
for tranche in tranches:
limits.append(float(slice.limit))
coeffr.append(float(tranche.coeffr))
coeffn.append(float(tranche.coeffn))
# add the constants
admindata.fromdict(constants.asdict())
# store admindata
self.__admindata = admindata
# return the value
return self.__admindata
except (IntegrityError, DatabaseError, InterfaceError) as error:
# Re-raise the exception in a different form
raise TaxError(27, f"{error}")
finally:
# close the session
if session:
session.close()
Notes
- line 11: the [ImpotsDaoWithAdminDataInDatabase] class inherits from the [AbstractImpôtsDao] class presented in version 4. We know that the latter implements the [InterfaceDao] interface presented in that same version. It is compliance with this interface that allows us to keep the [business] 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 [taxpayersdata.txt] file containing taxpayer data;
- [write_taxpayers_results] writes the results to the JSON file [results.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 [get_admindata] method 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: retrieve the [sqlalchemy] session that was created during application configuration by [config_database];
- line 40: we retrieve the tax brackets in a list;
- lines 43: we retrieve the constants for tax calculation;
- line 46: we create an instance of the [AdminData] class. Recall that it derives from [BaseEntity];
- lines 48–54: we initialize the arrays [limites, coeffr, coeffn] of the [AdminData] instance;
- lines 55–56: initialize the other properties of [AdminData] with the tax calculation constants. We took care to give the same names to the properties of the [AdminData] and [Constantes] classes, 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: error handling;
- lines 64–67: the database is only queried once. We can therefore close the [sqlalchemy] session;
20.2.4. Testing the [dao] layer
In version 4 of this application, we created a test class for the [business] layer. More specifically, it tested both the [business] and [DAO] layers. We are reusing this test to verify that the [DAO] layer is functioning as expected. The [business] layer, however, 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, 'surcharge': 0, 'discount': 0, 'reduction': 0, 'rate': 0.14}
taxpayer = TaxPayer().fromdict({"married": "yes", "children": 2, "salary": 55555})
job.calculate_tax(taxpayer, admindata)
# verification
self.assertAlmostEqual(taxpayer.tax, 2815, delta=1)
self.assertEqual(taxpayer.discount, 0)
self.assertEqual(taxpayer.reduction, 0)
self.assertAlmostEqual(taxpayer.rate, 0.14, delta=0.01)
self.assertEqual(taxpayer.surcharge, 0)
…
def test_11(self) -> None:
from TaxPayer import TaxPayer
# {'married': 'yes', 'children': 3, 'salary': 200000,
# 'tax': 42842, 'surcharge': 17283, 'discount': 0, 'reduction': 0, 'rate': 0.41}
taxpayer = TaxPayer().fromdict({'married': 'yes', 'children': 3, 'salary': 200000})
job.calculate_tax(taxpayer, admindata)
# checks
self.assertAlmostEqual(taxpayer.tax, 42842, 1)
self.assertEqual(taxpayer.discount, 0)
self.assertEqual(taxpayer.reduction, 0)
self.assertAlmostEqual(taxpayer.rate, 0.41, delta=0.01)
self.assertAlmostEqual(taxpayer.surcharge, 17283, delta=1)
if __name__ == '__main__':
# Expects a mysql or pgres parameter
import sys
syntax = f"{sys.argv[0]} mysql / pgres"
error = len(sys.argv) != 2
if not error:
dbms = sys.argv[1].lower()
error = dbsystem != "mysql" and dbsystem != "pgres"
if error:
print(f"syntax: {syntax}")
sys.exit()
# configure the application
import config
config = config.configure({'db': db})
# business logic
business = config['business']
try:
# admindata
admindata = config['dao'].get_admindata()
except BaseException as ex:
# display
print((f"The following error occurred: {ex}"))
# end
sys.exit()
# remove the parameter received by the script
sys.argv.pop()
# run the test methods
print("Tests in progress...")
unittest.main()
- We will not revisit the 11 tests described in the section |[business] layer test version 4|;
- lines 37–66: we will run the test script as a normal application rather than as a UnitTest. Line 66 is what will trigger the UnitTest framework. In the previous tests, we used the [setUp] method to configure the execution of each test. We were repeating 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 [business] on line 53 and [admindata] on line 56, which will then be used by the methods of [TestDaoBusiness], for example on line 12;
- lines 39–47: the test script expects a [mysql / pgres] parameter indicating whether a MySQL or PostgreSQL database is being used;
- lines 50–51: the test is configured;
- line 53: the [business] layer is retrieved from the configuration;
- line 56: we do the same with the [dao] layer. We then retrieve the [admindata] instance, which encapsulates the data needed to calculate the tax;
- Tests showed that the [unittest.main()] method on line 66 did not ignore the [mysql / pgres] parameter passed to the script, but instead 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 in progress...
...........
----------------------------------------------------------------------
Ran 11 tests in 0.001s
OK
Process finished with exit code 0
- Lines 5 and 7: all 11 tests passed;
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:
# we expect a mysql or pgres parameter
import sys
syntax = f"{sys.argv[0]} mysql / pgres"
error = len(sys.argv) != 2
if not error:
dbms = sys.argv[1].lower()
error = dbsystem != "mysql" and dbsystem != "pgres"
if error:
print(f"syntax: {syntax}")
sys.exit()
# configure the application
import config
config = config.configure({'db': db})
# The syspath is set—we can now import modules
from ImpôtsError import ImpôtsError
# Retrieve the application layers (they are already instantiated)
dao = config["dao"]
business = config["business"]
try:
# Retrieve tax brackets
admindata = dao.get_admindata()
# read taxpayer data
taxpayers = dao.get_taxpayers_data()["taxpayers"]
# any taxpayers?
if not taxpayers:
raise TaxError(57, f"No valid taxpayers in the file {config['taxpayersFilename']}")
# Calculate taxpayers' taxes
for taxPayer in taxpayers:
# taxPayer is both an input and output parameter
# taxPayer will be modified
business.calculate_tax(taxPayer, admindata)
# writing the results to a text file
dao.write_taxpayers_results(taxpayers)
except ImpôtsError as error:
# display the error
print(f"The following error occurred: {error}")
finally:
# Done
print("Work completed...")
Notes
- lines 1-10: we retrieve the [mysql / pgres] parameter, which specifies the DBMS 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: we retrieve references to the application layers;
- line 25: we request the tax administration data from the [dao] layer. The [business] layer needs this to calculate the tax;
- line 27: we retrieve the taxpayers’ data (id, married, 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 JSON file [results.json];
- lines 38–40: handle any errors;
To run the script, we create two |execution configurations|:
The results obtained in the [results.json] file are those from version 4.
20.3. Application 3: Calculating Tax 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 [main] script initiates the user dialogue using the [ui.run] method of the [ui] layer;
- 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 the [business] layer to perform this calculation;
The [config_layers] file instantiates an additional layer:
def configure(config: dict) -> dict:
# instantiate dao layer
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = TaxDaoWithAdminDataInDatabase(config)
# instantiate the [business] layer
from BusinessTaxes import BusinessTaxes
config['business'] = BusinessTaxes()
# UI
from TaxConsole import TaxConsole
config['ui'] = TaxConsole(config)
# 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:
# expect a mysql or pgres parameter
import sys
syntax = f"{sys.argv[0]} mysql / pgres"
error = len(sys.argv) != 2
if not error:
dbms = sys.argv[1].lower()
error = dbsystem != "mysql" and dbsystem != "pgres"
if error:
print(f"syntax: {syntax}")
sys.exit()
# configure the application
import config
config = config.configure({'db': db})
# The syspath is configured—we can now import modules
from ImpôtsError import ImpôtsError
# retrieve the [ui] layer
ui = config["ui"]
# code
try:
# run the [ui] layer
ui.run()
except ImpôtsError as ex1:
# display the error message
print(f"The following error occurred: {ex1}")
except BaseException as ex2:
# display the error message
print(f"The following error occurred: {ex2}")
finally:
# executed in all cases
print("Job completed...")
- lines 1-10: the script expects a parameter [mysql / pgres] specifying the DBMS 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|. It could not be otherwise since all the interfaces from version 4 have been preserved in version 5.