Introduction à Python 3 et au framework web Flask par l'exemple

20. Exercice d'application : version 5▲

Nous allons développer trois applications :

l’application 1 initialisera la base de données qui va venir remplacer le fichier [admindata.json] de la version 4 ;
l’application 2 fera le calcul des impôts en mode batch ;
l’application 3 fera le calcul des impôts en mode interactif ;

20-1. Application 1 : initialisation de la base de données▲

L’application 1 aura l'architecture suivante :

C'est une évolution de l'architecture de la version 4 (paragraphe Version 4) : les données fiscales seront trouvées dans une base de données au lieu d'être dans un fichier jSON. La couche [dao] va évoluer pour implémenter ce changement.

20-1-1. Le fichier [admindata.json]▲

Le fichier [admindata.json] est celui qu’il était dans la version 4 :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.

{
    "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
}

Nous allons utiliser comme colonnes de la base de données les clés de ce dictionnaire.

20-1-2. Création des bases de données▲

Comme il a été montré au paragraphe création d’une base de données MySQL, nous créons une base de données MySQL nommée [dbimpots-2019] propriété de l’utilisateur [admimpots] de mot de passe [mdpimpots]. Dans [phpMyAdmin] cela donne la chose suivante :

De même, comme il a été montré au paragraphe création d’une base de données Postgre SQL, nous créons une base de données PostgreSQL nommée [dbimpots-2019] propriété de l’utilisateur [admimpots] de mot de passe [mdpimpots]. Dans [pgAdmin] cela donne la chose suivante :

Les bases sont créées mais pour l’instant elles n’ont aucune table. Celles-ci vont être construites par l’ORM [sqlalchemy].

20-1-3. Les entités mappées par [sqlalchemy]▲

Nous allons créer deux tables pour encapsuler les données de [admindata.json] :

Définie par [sqlalchemy] la table [tbtranches] rassemblera les données des tableaux [limites, coeffr, coeffn] du dictionnaire [admindata.json] :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.

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

Définie par [sqlalchemy] la table [tbconstantes] rassemblera les constantes du dictionnaire [admindata.json] :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.

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

Les entités qui seront mappées avec ces deux tables seront les suivantes :

L’entité [Constantes] encapsule les constantes du dictionnaire [admindata.json] :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.

from BaseEntity import BaseEntity


# classe conteneur des données de l'administration fiscale
class Constantes(BaseEntity):
    # clés exclues de l'état de la classe
    excluded_keys = ["_sa_instance_state"]

    # clés autorisées
    @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"]

ligne 5 : la classe [Constantes] étend la classe [BaseEntity] ;
ligne 7 : par mapping [sqlalchemy], la classe [Constante] va recevoir la propriété [_sa_instance_state]. Nous l’excluons du dictionnaire [asdict] de l’entité ;
lignes 11-23 : les propriétés de l’entité. On a repris les noms utilisés dans le dictionnaire [admindata.json] pour faciliter l’écriture du code ;

L’entité [Tranche] encapsule une ligne des trois tableaux [limites, coeffr, coeffn] du dictionnaire [admindata.json] :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.

from BaseEntity import BaseEntity


# classe conteneur des données de l'administration fiscale
class Tranche(BaseEntity):
    # clés exclues de l'état de la classe
    excluded_keys = ["_sa_instance_state"]

    # clés autorisées
    @staticmethod
    def get_allowed_keys() -> list:
        return ["id", "limite", "coeffr", "coeffn"]

ligne 5 : la classe [Tranche] étend la classe [BaseEntity] ;
ligne 7 : on exclut des propriétés du dictionnaire [asdict] de l’entité, la propriété [_sa_instance_state] ajoutée par [sqlalchemy] ;
lignes 10-12 : les propriétés de la classe ;

Le mapping entre les entités [Constantes, Tranche] et les tables [constantes, tranches] sera le suivant :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.

…
    # 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)
                             )

    # 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)
                           )
    # les mappings
    from Tranche import Tranche
    mapper(Tranche, tranches_table)

    from Constantes import Constantes
    mapper(Constantes, constantes_table)

les mappings sont faits aux lignes 24-29. On y a omis de faire les correspondances entre propriétés des entités mappées et tables de la base de données. C’est possible lorsque les noms des colonnes des tables sont les mêmes que ceux des propriétés auxquelles elles doivent être associées. Pour cette raison, nous avons repris dans les tables les noms des propriétés des entités mappées. Cela facilite l’écriture du code et sa compréhension ;

20-1-4. Le fichier de configuration de [sqlalchemy]▲

Nous venons de détailler une partie de la configuration de [sqlalchemy]. Le fichier [config_database] dans sa totalité est le suivant :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.

def configure(config: dict) -> dict:
    # configuration sqlalchemy
    from sqlalchemy import create_engine, Table, Column, Integer, MetaData, Float
    from sqlalchemy.orm import mapper, sessionmaker

    # chaînes de connexion aux bases de données exploitées
    connection_strings = {
        'mysql': "mysql+mysqlconnector://admimpots:mdpimpots@localhost/dbimpots-2019",
        'pgres': "postgresql+psycopg2://admimpots:mdpimpots@localhost/dbimpots-2019"
    }
    # chaîne de connexion à la base de données exploitée
    engine = create_engine(connection_strings[config['sgbd']])

    # metadata
    metadata = MetaData()

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

    # 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)
                           )
    # les mappings
    from Tranche import Tranche
    mapper(Tranche, tranches_table)

    from Constantes import Constantes
    mapper(Constantes, constantes_table)

    # la session factory
    session_factory = sessionmaker()
    session_factory.configure(bind=engine)

    # une session
    session = session_factory()

    # on enregistre certaines informations
    config['database'] = {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
                          "constantes_table": constantes_table, "session": session}

    # résultat
    return config

ligne 1 : la fonction [configure] reçoit en paramètre un dictionnaire dont la clé [sgbd] lui dit quel SGBD utiliser : MySQL (mysql) ou PostgreSQL (pgres) ;
lignes 6-12 : on sélectionne la base de données demandée par la configuration ;
lignes 14-44 : mappings entités / tables. Ces mappings sont simples car il n’existe aucun lien entre les tables [tranches] et [constantes]. Elles sont indépendantes. Il n’y a donc pas de clé étrangère de l’une sur l’autre à gérer ;
lignes 46-51 : on crée la session [session] de travail de l’application ;
lignes 53-58 : les informations utiles sont mises dans le dictionnaire de la configuration et celui-ci est retourné ;

20-1-5. La couche [dao]▲

Revenons à l’architecture de l’application 1 à construire :

La couche [dao] [1] doit lire le fichier [admindata.json] [2] et transférer son contenu dans une des bases [3, 4] ;

La couche [dao] présente l’interface [1] et est implémentée par la classe [2].

L’interface [InterfaceDao4TransferAdminData2Database] est la suivante :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.

# imports
from abc import ABC, abstractmethod


# interface InterfaceImpôtsUI
class InterfaceDao4TransferAdminData2Database(ABC):
    # transfert des données fiscales dans une base de données
    @abstractmethod
    def transfer_admindata_in_database(self:object):
        pass

lignes 8-10 : l’interface ne présente qu’une méthode [transfer_admindata_in_database] sans paramètres. Comme cette méthode a besoin de paramètres (quel fichier ?, quelle base de données ?), cela signifie que ceux-ci seront passés au constructeur des classes implémentant cette interface ;

La classe [DaoTransferAdminDataFromJsonFile2Database] implémente l’interface [InterfaceDao4TransferAdminData2Database] de la façon suivante :

Sélectionnez

# 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):

    # constructeur
    def __init__(self, config: dict):
        self.config = config

    # transfert
    def transfer_admindata_in_database(self) -> None:
        # initialisations
        session = None
        config = self.config

        try:
            # on récupère les données de l'administration fiscale
            with codecs.open(config["admindataFilename"], "r", "utf8") as fd:
                # transfert du contenu dans un dictionnaire
                admindata = json.load(fd)

            # on récupère la configuration de la base de données
            database = config["database"]

            # suppression des deux tables de la base de données
            # checkfirst=True : vérifie d'abord que la table existe
            database["tranches_table"].drop(database["engine"], checkfirst=True)
            database["constantes_table"].drop(database["engine"], checkfirst=True)

            # recréation des tables à partir des mappings
            database["metadata"].create_all(database["engine"])

            # la session [sqlalchemy] courante
            session = database["session"]

            # on remplit la table des tranches de l'impôt
            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]}))
            # on remplit la table des constantes
            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"]
            }))

            # validation de la session [sqlalchemy]
            session.commit()
        except (IntegrityError, DatabaseError, InterfaceError) as erreur:
            # on relance l'exception sous une autre forme
            raise ImpôtsError(17, f"{erreur}")
        finally:
            # on libère les ressources de la session
            if session:
                session.close()

ligne 13 : la classe [DaoTransferAdminDataFromJsonFile2Database] implémente l’interface [InterfaceDao4TransferAdminData2Database] ;
lignes 15-17 : le constructeur de la classe reçoit en paramètre le dictionnaire de la configuration. Les clés suivantes vont être utilisées :
- [admindataFilename] (ligne 27) : le nom du fichier jSON contenant les données de l’administration fiscale à transférer en base ;
- [database] ligne 32 : le configuration [sqlalchemy] de l’application ;
lignes 34-37 : suppression des tables [constantes] et [tranches] si elles existent ;
lignes 39-40 : recréation des deux tables ;
ligne 43 : on récupère la session [sqlalchemy] présente dans la configuration ;
lignes 45-51 : les tableaux [limites, coeffr, coeffn] du dictionnaire [admindata] sont mis dans la session. Pour cela on met dans la session des instances de l’entité [Tranche] ;
lignes 52-64 : une instance de l’entité [Constantes] est mise en session ;
lignes 66-67 : la session est validée. Si les données de la session n’étaient pas encore en base, elles y sont mises à ce moment là ;
lignes 68-70 : gestion d’une éventuelle erreur ;
lignes 71-74 : la session est fermée. On peut le faire car la couche [dao] n’est utilisée qu’une fois ;

20-1-6. Configuration de l’application▲

L’application est configurée par trois fichiers [1] :

[config] est le fichier de configuration générale. C’est lui qui configure l’application [main]. Il se fait aider par les deux autres fichiers :
- [config_database] que nous avons étudié et qui configure l’ORM [sqlalchemy] ;
- [config_layers] qui configure les couches de l’application ;

Le fichier [config] est le suivant :

Sélectionnez

def configure(config: dict) -> dict:
    # [config] a la clé [sgbd] qui vaut:
    # [mysql] pour gérer une base MySQL
    # [pgres] pour gérer une base PostgreSQL

    import os

    # étape 1 ---
    # on établit le Python Path de l'application

    # chemin absolu du dossier de ce script
    script_dir = os.path.dirname(os.path.abspath(__file__))

    # root_dir (à changer éventuellement)
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"

    # chemins absolus des dépendances
    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",
        # dossiers locaux
        f"{script_dir}",
        f"{script_dir}/../../interfaces",
        f"{script_dir}/../../services",
        f"{script_dir}/../../entities",
    ]

    # on fixe le syspath
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    # é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"
    })

    # étape 3 ------
    # configuration base de données
    import config_database
    config = config_database.configure(config)

    # étape 4 ------
    # instanciation des couches de l'application
    import config_layers
    config = config_layers.configure(config)

    # on rend la config
    return config

lignes 8-36 : on construit le Python Path de l’application ;
lignes 38-43 : on met dans la configuration le chemin du fichier [admindata.json] ;
lignes 45-48 : configuration [sqlalchemy] ;
lignes 50-53 : instanciation des couches de l’application ;
ligne 56 : on rend la configuration générale ;

Le fichier [config_layers] est le suivant :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.

def configure(config: dict) -> dict:
    # instanciation couche [dao]
    from DaoTransferAdminDataFromJsonFile2Database import DaoTransferAdminDataFromJsonFile2Database
    config['dao'] = DaoTransferAdminDataFromJsonFile2Database(config)

    # on rend la config
    return config

lignes 3-4 : instanciation de la couche [dao]. On a vu que le constructeur de la classe [DaoTransferAdminDataFromJsonFile2Database] attendait en paramètre le dictionnaire de la configuration générale de l’application ;
ligne 4 : la référence sur la couche [dao] est mise dans la configuration ;
ligne 7 : on rend la configuration ;

20-1-7. Le script [main] de l’application▲

Le script principal [main] est le suivant :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.

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

# le syspath est établi - on peut faire les imports
from ImpôtsError import ImpôtsError

# on récupère la couche [dao]
dao = config["dao"]

# code
try:
    # transfert des données dans la base
    dao.transfer_admindata_in_database()
except ImpôtsError as ex1:
    # on affiche l'erreur
    print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
    # on affiche l'erreur
    print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
    # fin
    print("Terminé...")

lignes 1-10 : on attend un paramètre. On vérifie qu’il est là et correct ;
lignes 12-14 : on configure l’application (générale, sqlalchemy, couches) en passant en paramètre le type de SGBD choisi ;
lignes 19-20 : on va avoir besoin de la couche [dao]. On la récupère ;
ligne 25 : on fait le transfert en base. Toutes les informations nécessaires à la méthode [transfer_admindata_in_database] sont disponibles dans les propriétés de la couche [dao] de la ligne 20. C’est là qu’elle ira les chercher ;

Après exécution avec la base MySQL, celle-ci contient les éléments suivants (phpMyAdmin) :

Colonne [3], on voit les valeurs attribuées par MySQL à la clé primaire [id]. La numérotation démarre à 1. La copie d’écran ci-dessus a été obtenue après plusieurs exécutions du script.

Avec la base PostgreSQL les résultats sont les suivants :

on clique droit sur [1], puis ensuite [2-3] ;
en [4], on a bien les données des tranches d’impôts ;

On refait la même chose pour la table des constantes [tbconstantes] :

20-2. Application 2 : calcul de l’impôt en mode batch▲

20-2-1. Architecture▲

L’application de calcul de l’impôt de la version 4 utilisait l'architecture suivante :

La couche [dao] implémente une interface [InterfaceImpôtsDao]. Nous avons construit une classe implémentant cette interface :

[ImpôtsDaoWithAdminDataInJsonFile] qui allait chercher les données fiscales dans un fichier jSON. C’était la version 3 ;

Nous allons implémenter l’interface [InterfaceImpôtsDao] par une nouvelle classe [ImpotsDaoWithTaxAdminDataInDatabase] qui ira chercher les données de l’administration fiscale dans une base de données. La couche [dao], comme précédemment, écrira les résultats dans un fichier jSON et trouvera les données des contribuables dans un fichier texte. Nous savons que si nous continuons à respecter l’interface [InterfaceImpôtsDao], la couche [métier] n’aura pas à être modifiée.

La nouvelle architecture sera la suivante :

20-2-2. Configuration de l’application▲

Le fichier de configuration [config_database] reste ce qu’il était dans l’application 1. La configuration [config] intègre des éléments nouveaux :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.
8.
9.

    # é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"
    })

lignes 6-8 : les chemins absolus des fichiers texte utilisés par l’application 2 ;

La configuration des couches [config_layers] évolue de la façon suivante :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.

def configure(config: dict) -> dict:
    # instanciation couche dao
    from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
    config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)

    # instanciation couche [métier]
    from ImpôtsMétier import ImpôtsMétier
    config['métier'] = ImpôtsMétier()

    # on rend la config
    return config

lignes 3-4 : la couche [dao] est désormais implémentée par la classe [ImpotsDaoWithAdminDataInDatabase]. Cette classe est nouvelle mais implémente la même interface [InterfaceDao] que la version 4 de l’exercice d’application ;
lignes 7-8 : la couche [métier] est implémentée par la classe [ImpôtsMétier]. C’est la classe utilisée dans la version 4 de l’exercice d’application ;

20-2-3. La couche [dao]▲

La classe d'implémentation [ImpotsDaoWithAdminDataInDatabase] de l'interface [InterfaceImpôtsDao] sera la suivante :

Sélectionnez

# 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):
    # constructeur
    def __init__(self, config: dict):
        # config["taxPayersFilename"] : le nom du fichier texte des contribuables
        # config["taxPayersResultsFilename"] : le nom du fichier jSON des résultats
        # config["errorsFilename"] : enregistre les erreurs trouvées dans taxPayersFilename
        # config["database"] : configuration de la base de données

        # initialisation de la classe Parent
        AbstractImpôtsDao.__init__(self, config)
        # mémorisation paramètre
        self.__config = config
        # admindata
        self.__admindata = None

    # implémentation de l'interface
    def get_admindata(self):
        # admindata a-t-il été mémorisé ?
        if self.__admindata:
            return self.__admindata
        # on fait une requête en BD
        session = None
        config = self.__config
        try:
            # une session
            database_config = config["database"]
            session = database_config["session"]

            # on lit la table des tranches de l'impôt
            tranches = session.query(Tranche).all()

            # on lit la table des constantes (1 seule ligne)
            constantes = session.query(Constantes).first()

            # on crée l'instance admindata
            admindata = AdminData()
            # on y crée les tableaux limtes, 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))
            # on y rajoute les constantes
            admindata.fromdict(constantes.asdict())
            # on mémorise admindata
            self.__admindata = admindata
            # on rend la valeur
            return self.__admindata
        except (IntegrityError, DatabaseError, InterfaceError) as erreur:
            # on relance l'exception sous une autre forme
            raise ImpôtsError(27, f"{erreur}")
        finally:
            # on ferme la session
            if session:
                session.close()

Notes

ligne 11 : la classe [ImpotsDaoWithAdminDataInDatabase] hérite de la classe AbstractImpôtsDao présentée dans la version 4. On sait que cette dernière implémente l’interface InterfaceDao présentée dans cette même version. C’est le respect de cette interface qui nous permet de ne pas changer la couche [métier] ;
ligne 13 : le constructeur de la classe reçoit en paramètre le dictionnaire de la configuration de l’application ;
ligne 20 : la classe parent [] est initialisée. Elle implémente partiellement l’interface [InterfaceDao] :
- [get_taxpayers_data] lit le fichier [taxpayersdata.txt] qui contient les données des contribuables ;
- [write_taxpayers_results] écrit les résultats dans le fichier jSON [résultats.json] ;
- [get_admindata] n’est pas implémentée ;
ligne 22 : on mémorise la configuration passée en paramètres ;
ligne 27 : implémentation de la méthode [get_admindata] de l’interface [InterfaceDao] :
lignes 28-30 : la méthode [get_admindata] récupère les données de l’administration fiscale dans un objet de type [AdminData] et mémorise cet objet dans [self.__admindata]. Si la méthode [get_admindata] est appelée plusieurs fois, on n’interroge pas la base de données plusieurs fois. On l’interroge seulement la première fois. Les fois suivantes, on rend l’objet [self.__admindata] ;
lignes 36-37 : on récupère la session [sqlalchemy] qui a été créée lors de la configuration de l’application par [config_database] ;
lignes 40 : on récupère les tranches de l’impôt dans une liste ;
lignes 43 : on récupère les constantes du calcul de l’impôt ;
ligne 46 : on crée une instance de la classe AdminData. On rappelle qu’elle dérive de [BaseEntity] ;
lignes 48-54 : on initialise les tableaux [limites, coeffr, coeffn] de l’instance [AdminData] ;
lignes 55-56 : on initialise les autres propriétés de [AdminData] avec les constantes du calcul de l’impôt. On avait pris soin de donner les mêmes noms aux propriétés des classes [AdminData] et [Constantes], ce qui simplifie le code ;
lignes 57-58 : l’instance [AdminData] est mémorisée dans la couche [dao] pour la rendre lors des prochains appels à la méthode [get_admindata] ;
ligne 60 : on rend la valeur demandée par le code appelant ;
lignes 61-63 : gestion d’une éventuelle erreur ;
lignes 64-67 : la base de données ne fait l’objet que d’une unique requête. On peut donc fermer la session [sqlalchemy] ;

20-2-4. Test de la couche [dao]▲

Dans la version 4 de cette application, nous avions construit une classe de test de la couche [métier]. Plus exactement, elle testait à la fois les couches [métier] et [dao]. Nous reprenons ce test pour vérifier que la couche [dao] fonctionne comme attendu. En effet, la couche [métier] elle ne change pas.

Le test [TestDaoMétier] est le suivant :

Sélectionnez

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

nous ne revenons pas sur les 11 tests décrits au paragraphe test couche [métier] version 4 ;
lignes 37-66 : nous allons exécuter le script des tests comme une application normale et non pas comme un test UnitTest. C’est la ligne66qui fera intervenir le framework UnitTest. Dans les tests précédents, nous utilisions la méthode [setUp] pour configurer l’exécution de chaque test. On refaisait 11 fois la même configuration puisque la fonction [setUp] est exécutée avant chaque test. Ici, nous faisons la configuration 1 fois. Elle consiste à définir des variables globales [métier] ligne 53, [admindata] ligne 56 qui seront ensuite utilisées par les méthodes de [TestDaoMétier], ligne 12 par exemple ;
lignes 39-47 : le script de test attend un paramètre [mysql / pgres] qui indique si on utilise une base MySQL ou PostgreSQL ;
lignes 50-51 : le test est configuré ;
ligne 53 : on récupère la couche [métier] dans la configuration ;
ligne 56 : on fait de même avec la couche [dao]. On récupère alors l’instance [admindata] qui encapsule les données nécessaires au calcul de l’impôt ;
les tests ont montré que la méthode [unittest.main()] de la ligne 66 n’ignorait pas le paramètre [mysql / pgres] reçu par le script mais lui donnait une signification autre. La ligne 63 fait en sorte que cette méthode n’ait plus aucun paramètre ;

Nous créons deux configurations d’exécution :

Si nous exécutons l’une de ces deux configurations, nous obtenons les résultats suivants :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.
8.
9.

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

lignes 5 et 7 : les 11 tests ont été réussis ;

Rappelons que ces tests ne vérifient que 11 cas du calcul de l’impôt. Leur réussite peut néanmoins suffire pour nous donner confiance dans la couche [dao].

20-2-5. Le script principal▲

Le script principal [main] est le même que dans la version 4 :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.

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

# le syspath est établi - on peut faire les imports
from ImpôtsError import ImpôtsError

# on récupère les couches de l'application (elles sont déjà instanciées)
dao = config["dao"]
métier = config["métier"]

try:
    # récupération des tranches de l'impôt
    admindata = dao.get_admindata()
    # lecture des données des contribuables
    taxpayers = dao.get_taxpayers_data()["taxpayers"]
    # des contribuables ?
    if not taxpayers:
        raise ImpôtsError(57, f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
    # calcul de l'impôt des contribuables
    for taxPayer in taxpayers:
        # taxPayer est à la fois un paramètre d'entrée et de sortei
        # taxPayer va être modifié
        métier.calculate_tax(taxPayer, admindata)
    # écriture des résultats dans un fichier texte
    dao.write_taxpayers_results(taxpayers)
except ImpôtsError as erreur:
    # affichage de l'erreur
    print(f"L'erreur suivante s'est produite : {erreur}")
finally:
    # terminé
    print("Travail terminé...")

Notes

lignes 1-10 : on récupère le paramètre [mysql / pgres] qui indique le SGBD à utiliser ;
lignes 12-14 : l’application est configurée ;
lignes 16-17 : la classe [ImpôtsError] est importée. On en a besoin ligne 38 ;
lignes 19-21 : on récupère des références sur les couches de l’application ;
ligne 25 : on demande à la couche [dao] les données de l’administration fiscale. La couche [métier] en a besoin pour le calcul de l’impôt ;
ligne 27 : on récupère dans une liste, les données (id, marié, enfants, salaire) des contribuables ;
lignes 29-30 : si cette liste est vide, on lance une exception ;
lignes 32-35 : calcul de l'impôt des éléments de la liste [taxpayers] ;
ligne 37 : écriture des résultats dans le fichier jSON[résultats.json] ;
lignes 38-40 : gestion de l'éventuelle erreur ;

Pour l’exécution du script, on crée deux configurations d’exécution :

Les résultats obtenus dans le fichier [résultats.json] sont ceux de la version 4.

20-3. Application 3 : calcul de l’impôt en mode interactif▲

Nous introduisons maintenant l’application permettant de calculer l’impôt de façon interactive. C’est un portage de l’application 2 de la version 4.

le script [main] lance le dialogue avec l’utilisateur avec la méthode [ui.run] de la couche [ui] ;
la couche [ui] :
- utilise la couche [dao] pour obtenir les données permettant de faire le calcul de l’impôt ;
- demande à l’utilisateur les renseignements concernant le contribuable dont on veut calculer l’impôt ;
- utilise la couche [métier] pour faire ce calcul ;

Le fichier [config_layers] instancie une couche supplémentaire :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.

def configure(config: dict) -> dict:
    # instanciation couche dao
    from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
    config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)

    # instanciation couche [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)

    # on rend la config
    return config

La classe [ImpôtsConsole], lignes 11-12, est la même que dans la version 4.

Le script principal [main] est le suivant :

Sélectionnez

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.

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

# le syspath est configuré - on peut faire les imports
from ImpôtsError import ImpôtsError

# on récupère la couche [ui]
ui = config["ui"]

# code
try:
    # exécution de la couche [ui]
    ui.run()
except ImpôtsError as ex1:
    # on affiche le message d'erreur
    print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
    # on affiche le message d'erreur
    print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
    # exécuté dans tous les cas
    print("Travail terminé...")

lignes 1-10, le script attend un paramètre [mysql / pgres] qui indique le SGBD à utiliser ;
lignes 12-14 : l’application est configurée ;
lignes 19-20 : on récupère la couche [ui] dans la configuration ;
ligne 25 : on l’exécute ;

Les résultats sont identiques à ceux de la version 4. Il ne pouvait en être autrement puisque toutes les interfaces de la version 4 ont été respectées dans la version 5.