Skip to content

20. Esercizio pratico: versione 5

Image

Svilupperemo tre applicazioni:

  • L'applicazione 1 inizializzerà il database che sostituirà il file [admindata.json] della versione 4;
  • L'applicazione 2 calcolerà le imposte in modalità batch;
  • L'applicazione 3 calcolerà le imposte in modalità interattiva;

20.1. Applicazione 1: Inizializzazione del database

L'applicazione 1 avrà la seguente architettura:

Image

Si tratta di un'evoluzione dell'architettura della versione 4 (vedere la sezione |Versione 4|): i dati fiscali saranno memorizzati in un database anziché in un file JSON. Il livello [DAO] sarà aggiornato per implementare questa modifica.

20.1.1. Il file [admindata.json]

Image

Il file [admindata.json] è lo stesso della versione 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
}

Useremo le chiavi di questo dizionario come colonne nel database.

20.1.2. Creazione dei database

Come mostrato nella sezione |creazione di un database MySQL|, creiamo un database MySQL denominato [dbimpots-2019] di proprietà dell'utente [admimpots] con password [mdpimpots]. In [phpMyAdmin], appare come segue:

Image

Allo stesso modo, come illustrato nella sezione |Creazione di un database PostgreSQL|, creiamo un database PostgreSQL denominato [dbimpots-2019] di proprietà dell'utente [admimpots] con la password [mdpimpots]. In [pgAdmin], la situazione si presenta come segue:

Image

I database sono stati creati, ma per ora non contengono tabelle. Queste saranno generate dall'ORM [sqlalchemy].

20.1.3. Entità mappate da [sqlalchemy]

Creeremo due tabelle per incapsulare i dati da [admindata.json]:

Definita da [sqlalchemy], la tabella [tbtranches] raccoglierà i dati dagli array [limites, coeffr, coeffn] nel dizionario [admindata.json]:


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

Definita da [sqlalchemy], la tabella [tbconstantes] conterrà le costanti del dizionario [admindata.json]:


    # la table des constantes
    constantes_table = Table("tbconstantes", metadata,
                             Column('id', Integer, primary_key=True),
                             Column('plafond_qf_demi_part', Float, nullable=False),
                             Column('plafond_revenus_celibataire_pour_reduction', Float, nullable=False),
                             Column('plafond_revenus_couple_pour_reduction', Float, nullable=False),
                             Column('valeur_reduc_demi_part', Float, nullable=False),
                             Column('plafond_decote_celibataire', Float, nullable=False),
                             Column('plafond_decote_couple', Float, nullable=False),
                             Column('plafond_impot_celibataire_pour_decote', Float, nullable=False),
                             Column('plafond_impot_couple_pour_decote', Float, nullable=False),
                             Column('abattement_dixpourcent_max', Float, nullable=False),
                             Column('abattement_dixpourcent_min', Float, nullable=False)
                             )

Le entità che verranno mappate su queste due tabelle sono le seguenti:

Image

L'entità [Constants] incapsula le costanti del dizionario [admindata.json]:

 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


#  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"]
  • riga 5: la classe [Constants] estende la classe [BaseEntity];
  • riga 7: tramite il mapping [sqlalchemy], la classe [Constante] riceverà la proprietà [_sa_instance_state]. La escludiamo dal dizionario [asdict] dell’entità;
  • righe 11–23: le proprietà dell'entità. Abbiamo riutilizzato i nomi utilizzati nel dizionario [admindata.json] per facilitare la scrittura del codice;

L'entità [Tranche] incapsula una riga dei tre array [limites, coeffr, coeffn] presenti nel dizionario [admindata.json]:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
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"]
  • riga 5: la classe [Tranche] estende la classe [BaseEntity];
  • riga 7: la proprietà [_sa_instance_state] aggiunta da [sqlalchemy] è esclusa dal dizionario [asdict] dell'entità;
  • righe 10–12: le proprietà della classe;

La mappatura tra le entità [Constants, Slice] e le tabelle [constants, slices] sarà la seguente:

Image

 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
…
    #  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)
  • Le mappature sono definite alle righe 24–29. Abbiamo omesso le mappature tra le proprietà delle entità mappate e le tabelle del database. Ciò è possibile quando i nomi delle colonne delle tabelle coincidono con quelli delle proprietà a cui devono essere associate. Per questo motivo, abbiamo incluso i nomi delle proprietà delle entità mappate nelle tabelle. Ciò rende il codice più facile da scrivere e da comprendere;

20.1.4. Il file di configurazione [sqlalchemy]

Image

Abbiamo appena descritto in dettaglio una parte della configurazione [sqlalchemy]. Il file [config_database] completo è il seguente:

 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:
    #  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
  • riga 1: la funzione [configure] riceve come parametro un dizionario, la cui chiave [dbms] indica quale DBMS utilizzare: MySQL (mysql) o PostgreSQL (pgres);
  • righe 6–12: viene selezionato il database specificato dalla configurazione;
  • righe 14–44: mappature entità/tabella. Queste mappature sono semplici perché non c'è alcuna relazione tra le tabelle [tranches] e [constantes]. Sono indipendenti. Pertanto, non ci sono chiavi esterne tra loro da gestire;
  • righe 46–51: creazione della sessione di lavoro dell'applicazione [session];
  • righe 53–58: le informazioni rilevanti vengono inserite nel dizionario di configurazione, che viene poi restituito;

20.1.5. Il livello [dao]

Torniamo all'architettura dell'Applicazione 1 da realizzare:

Image

Il livello [dao] [1] deve leggere il file [admindata.json] [2] e trasferirne il contenuto a uno dei database [3, 4];

Image

Il livello [dao] fornisce l'interfaccia [1] ed è implementato dalla classe [2].

L'interfaccia [InterfaceDao4TransferAdminData2Database] è la seguente:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
#  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
  • righe 8–10: l'interfaccia definisce un solo metodo [transfer_admindata_in_database] senza parametri. Poiché questo metodo richiede dei parametri (quale file?, quale database?), ciò significa che tali parametri saranno passati al costruttore delle classi che implementano questa interfaccia;

La classe [DaoTransferAdminDataFromJsonFile2Database] implementa l'interfaccia [InterfaceDao4TransferAdminData2Database] come segue:

 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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
#  imports
import codecs
import json

from sqlalchemy.exc import DatabaseError, IntegrityError, InterfaceError

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


class DaoTransferAdminDataFromJsonFile2Database(InterfaceDao4TransferAdminData2Database):

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

    #  transfer
    def transfer_admindata_in_database(self) -> None:
        #  initializations
        session = None
        config = self.config

        try:
            #  we retrieve data from the tax authorities
            with codecs.open(config["admindataFilename"], "r", "utf8") as fd:
                #  transfer content to a dictionary
                admindata = json.load(fd)

            #  retrieve the database configuration
            database = config["database"]

            #  delete the two tables from the database
            #  checkfirst=True: first checks that the table exists
            database["tranches_table"].drop(database["engine"], checkfirst=True)
            database["constantes_table"].drop(database["engine"], checkfirst=True)

            #  recreate tables from mappings
            database["metadata"].create_all(database["engine"])

            #  the current [sqlalchemy] session
            session = database["session"]

            #  fill in the tax bracket table
            limites = admindata["limites"]
            coeffr = admindata["coeffr"]
            coeffn = admindata["coeffn"]
            for i in range(len(limites)):
                session.add(Tranche().fromdict(
                    {"limite": limites[i], "coeffr": coeffr[i], "coeffn": coeffn[i]}))
            #  fill in the constants table
            session.add(Constantes().fromdict({
                'plafond_qf_demi_part': admindata["plafond_qf_demi_part"],
                'plafond_revenus_celibataire_pour_reduction': admindata["plafond_revenus_celibataire_pour_reduction"],
                'plafond_revenus_couple_pour_reduction': admindata["plafond_revenus_couple_pour_reduction"],
                'valeur_reduc_demi_part': admindata["valeur_reduc_demi_part"],
                'plafond_decote_celibataire': admindata["plafond_decote_celibataire"],
                'plafond_decote_couple': admindata["plafond_decote_couple"],
                'plafond_impot_celibataire_pour_decote': admindata["plafond_impot_celibataire_pour_decote"],
                'plafond_impot_couple_pour_decote': admindata["plafond_impot_couple_pour_decote"],
                'abattement_dixpourcent_max': admindata["abattement_dixpourcent_max"],
                'abattement_dixpourcent_min': admindata["abattement_dixpourcent_min"]
            }))

            #  session validation [sqlalchemy]
            session.commit()
        except (IntegrityError, DatabaseError, InterfaceError) as erreur:
            #  we relaunch the exception in another form
            raise ImpôtsError(17, f"{erreur}")
        finally:
            #  session resources are released
            if session:
                session.close()
  • riga 13: la classe [DaoTransferAdminDataFromJsonFile2Database] implementa l'interfaccia [InterfaceDao4TransferAdminData2Database];
  • righe 15–17: il costruttore della classe accetta il dizionario di configurazione come parametro. Verranno utilizzate le seguenti chiavi:
    • [admindataFilename] (riga 27): il nome del file JSON contenente i dati dell'amministrazione fiscale da trasferire al database;
    • [database] riga 32: la configurazione [sqlalchemy] dell’applicazione;
  • righe 34–37: eliminazione delle tabelle [constants] e [brackets] se presenti;
  • righe 39–40: ricreazione delle due tabelle;
  • riga 43: recupero della sessione [sqlalchemy] dalla configurazione;
  • righe 45–51: gli array [limits, coeffr, coeffn] del dizionario [admindata] vengono aggiunti alla sessione. A tal fine, vengono aggiunte alla sessione delle istanze dell'entità [Tranche];
  • righe 52–64: un'istanza dell'entità [Constantes] viene aggiunta alla sessione;
  • righe 66–67: la sessione viene convalidata. Se i dati della sessione non erano ancora presenti nel database, vengono inseriti a questo punto;
  • righe 68–70: gestione degli errori;
  • righe 71–74: la sessione viene chiusa. Ciò è possibile perché il livello [dao] viene utilizzato una sola volta;

20.1.6. Configurazione dell'applicazione

Image

L'applicazione è configurata da tre file [1]:

  • [config] è il file di configurazione generale. Configura l'applicazione [main]. È supportato dagli altri due file:
    • [config_database], che abbiamo già esaminato e che configura l'ORM [sqlalchemy];
    • [config_layers], che configura i livelli dell'applicazione;

Il file [config] è il seguente:

 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
def configure(config: dict) -> dict:
    #  [config] has the key [sgbd]:
    #  [mysql] to manage a MySQL database
    #  [pgres] to manage a PostgreSQL database

    import os

    #  step 1 ---
    #  establish the application's Python Path

    #  absolute path of this script's folder
    script_dir = os.path.dirname(os.path.abspath(__file__))

    #  root_dir (change if necessary)
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"

    #  absolute paths of dependencies
    absolute_dependencies = [
        #  InterfaceImpôtsDao, InterfaceImpôtsMétier, InterfaceImpôtsUi
        f"{root_dir}/impots/v04/interfaces",
        #  AbstractImpôtsDao, ImpôtsConsole, ImpôtsMétier
        f"{root_dir}/impots/v04/services",
        #  AdminData, ImpôtsError, TaxPayer
        f"{root_dir}/impots/v04/entities",
        #  BaseEntity, MyException
        f"{root_dir}/classes/02/entities",
        #  local files
        f"{script_dir}",
        f"{script_dir}/../../interfaces",
        f"{script_dir}/../../services",
        f"{script_dir}/../../entities",
    ]

    #  set the syspath
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    #  step 2 ------
    #  complete application configuration
    config.update({
        #  absolute paths for data files
        "admindataFilename": f"{script_dir}/../../data/input/admindata.json"
    })

    #  step 3 ------
    #  database configuration
    import config_database
    config = config_database.configure(config)

    #  step 4 ------
    #  instantiation of application layers
    import config_layers
    config = config_layers.configure(config)

    #  return the config
    return config
  • righe 8–36: Crea il Python Path dell'applicazione;
  • righe 38–43: aggiungere il percorso del file [admindata.json] alla configurazione;
  • righe 45–48: configurazione [SQLAlchemy];
  • righe 50–53: istanziare i livelli dell'applicazione;
  • riga 56: restituisce la configurazione generale;

Il file [config_layers] è il seguente:

1
2
3
4
5
6
7
def configure(config: dict) -> dict:
    #  layer instantiation [dao]
    from DaoTransferAdminDataFromJsonFile2Database import DaoTransferAdminDataFromJsonFile2Database
    config['dao'] = DaoTransferAdminDataFromJsonFile2Database(config)

    #  return the config
    return config
  • righe 3-4: istanziamento del livello [dao]. Abbiamo visto che il costruttore della classe [DaoTransferAdminDataFromJsonFile2Database] richiede come parametro il dizionario di configurazione generale dell'applicazione;
  • riga 4: il riferimento al livello [dao] viene aggiunto alla configurazione;
  • riga 7: restituisce la configurazione;

20.1.7. Lo script [main] dell'applicazione

Image

Image

Lo script principale [main] è il seguente:

 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
#  a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
    sgbd = sys.argv[1].lower()
    erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
    print(f"syntaxe : {syntaxe}")
    sys.exit()

#  configure the application
import config
config = config.configure({'sgbd': sgbd})

#  the syspath is set up - imports can be made
from ImpôtsError import ImpôtsError

#  we recover the [dao] layer
dao = config["dao"]

#  code
try:
    #  data transfer to the database
    dao.transfer_admindata_in_database()
except ImpôtsError as ex1:
    #  error is displayed
    print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
    #  error is displayed
    print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
    #  end
    print("Terminé...")
  • righe 1–10: attendiamo un parametro. Verifichiamo che sia presente e corretto;
  • righe 12–14: configuriamo l'applicazione (general, SQLAlchemy, layers) passando il tipo di DBMS scelto come parametro;
  • righe 19-20: avremo bisogno del livello [dao]. Lo recuperiamo;
  • riga 25: eseguiamo il trasferimento al database. Tutte le informazioni richieste dal metodo [transfer_admindata_in_database] sono disponibili nelle proprietà del livello [dao] della riga 20. È da lì che le recupererà;

Dopo aver eseguito lo script con il database MySQL, esso contiene i seguenti elementi (phpMyAdmin):

Image

Image

Image

La colonna [3] mostra i valori assegnati da MySQL alla chiave primaria [id]. La numerazione parte da 1. Lo screenshot sopra è stato acquisito dopo aver eseguito lo script diverse volte.

Image

Image

Con il database PostgreSQL, i risultati sono i seguenti:

Image

  • Fare clic con il tasto destro del mouse su [1], quindi su [2-3];
  • In [4] vengono visualizzati chiaramente i dati relativi alle fasce di imposta;

Facciamo lo stesso per la tabella delle costanti [tbconstantes]:

Image

Image

Image

20.2. Applicazione 2: Calcolo delle imposte in modalità batch

Image

20.2.1. Architettura

L'applicazione per il calcolo delle imposte nella versione 4 utilizzava la seguente architettura:

Image

Il livello [dao] implementa un'interfaccia [InterfaceImpôtsDao]. Abbiamo creato una classe che implementa questa interfaccia:

  • [TaxDaoWithAdminDataInJsonFile], che recuperava i dati fiscali da un file JSON. Quella era la versione 3;

Implementeremo l'interfaccia [InterfaceImpôtsDao] utilizzando una nuova classe [ImpotsDaoWithTaxAdminDataInDatabase] che recupererà i dati dell'amministrazione fiscale da un database. Il livello [dao], come in precedenza, scriverà i risultati in un file JSON e recupererà i dati dei contribuenti da un file di testo. Sappiamo che se continuiamo ad aderire all'interfaccia [InterfaceImpôtsDao], il livello [business] non dovrà essere modificato.

La nuova architettura sarà la seguente:

Image

20.2.2. Configurazione dell'applicazione

Image

Il file di configurazione [config_database] rimane lo stesso dell'Applicazione 1. La configurazione [config] include nuovi elementi:


    # é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"
    })
  • righe 6–8: i percorsi assoluti dei file di testo utilizzati dall'applicazione 2;

La configurazione dei livelli [config_layers] si evolve come segue:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
def configure(config: dict) -> dict:
    #  dao layer instantiation
    from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
    config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)

    #  instantiation layer [business]
    from ImpôtsMétier import ImpôtsMétier
    config['métier'] = ImpôtsMétier()

    #  return the config
    return config
  • righe 3-4: il livello [dao] è ora implementato dalla classe [TaxDaoWithAdminDataInDatabase]. Questa classe è nuova ma implementa la stessa interfaccia [DaoInterface] della versione 4 dell'esercizio dell'applicazione;
  • righe 7-8: il livello [business] è implementato dalla classe [ImpôtsMétier]. Questa è la classe utilizzata nella versione 4 dell'esercizio dell'applicazione;

20.2.3. Il livello [DAO]

La classe di implementazione [ImpotsDaoWithAdminDataInDatabase] per l'interfaccia [InterfaceImpôtsDao] sarà la seguente:

 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
59
60
61
62
63
64
65
66
67
#  imports
from sqlalchemy.exc import DatabaseError, IntegrityError, InterfaceError

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


class ImpotsDaoWithAdminDataInDatabase(AbstractImpôtsDao):
    #  manufacturer
    def __init__(self, config: dict):
        #  config["taxPayersFilename"]: name of the taxpayer text file
        #  config["taxPayersResultsFilename"]: name of the jSON results file
        #  config["errorsFilename"]: saves errors found in taxPayersFilename
        #  config["database"]: database configuration

        #  parent class initialization
        AbstractImpôtsDao.__init__(self, config)
        #  parameter memory
        self.__config = config
        #  admindata
        self.__admindata = None

    #  interface implementation
    def get_admindata(self):
        #  admindata memorized?
        if self.__admindata:
            return self.__admindata
        #  make a query in BD
        session = None
        config = self.__config
        try:
            #  a session
            database_config = config["database"]
            session = database_config["session"]

            #  read the table of tax brackets
            tranches = session.query(Tranche).all()

            #  read the constants table (1 line only)
            constantes = session.query(Constantes).first()

            #  create the admindata instance
            admindata = AdminData()
            #  we create limtes arrays, coeffR, coeffN
            limites = admindata.limites = []
            coeffr = admindata.coeffr = []
            coeffn = admindata.coeffn = []
            for tranche in tranches:
                limites.append(float(tranche.limite))
                coeffr.append(float(tranche.coeffr))
                coeffn.append(float(tranche.coeffn))
            #  we add the constants
            admindata.fromdict(constantes.asdict())
            #  admindata is memorized
            self.__admindata = admindata
            #  we return the value
            return self.__admindata
        except (IntegrityError, DatabaseError, InterfaceError) as erreur:
            #  we relaunch the exception in another form
            raise ImpôtsError(27, f"{erreur}")
        finally:
            #  close session
            if session:
                session.close()

Note

  • riga 11: la classe [ImpotsDaoWithAdminDataInDatabase] eredita dalla classe [AbstractImpôtsDao] presentata nella versione 4. Sappiamo che quest'ultima implementa l'interfaccia [InterfaceDao] presentata nella stessa versione. È proprio la conformità a questa interfaccia che ci permette di mantenere invariato il livello [business];
  • riga 13: il costruttore della classe riceve come parametro il dizionario di configurazione dell'applicazione;
  • riga 20: la classe padre [] viene inizializzata. Essa implementa parzialmente l'interfaccia [InterfaceDao]:
    • [get_taxpayers_data] legge il file [taxpayersdata.txt] contenente i dati dei contribuenti;
    • [write_taxpayers_results] scrive i risultati nel file JSON [results.json];
    • [get_admindata] non è implementato;
  • riga 22: la configurazione passata come parametri viene memorizzata;
  • riga 27: implementazione del metodo [get_admindata] dell'interfaccia [InterfaceDao]:
  • righe 28–30: il metodo [get_admindata] recupera i dati dall'amministrazione fiscale in un oggetto di tipo [AdminData] e memorizza questo oggetto in [self.__admindata]. Se il metodo [get_admindata] viene chiamato più volte, il database non viene interrogato più volte. Viene interrogato solo la prima volta. Nelle chiamate successive, viene restituito l'oggetto [self.__admindata];
  • righe 36–37: recuperiamo la sessione [sqlalchemy] creata durante la configurazione dell'applicazione da [config_database];
  • riga 40: recuperiamo le fasce di imposta in un elenco;
  • riga 43: recuperiamo le costanti per il calcolo delle imposte;
  • riga 46: creiamo un'istanza della classe [AdminData]. Ricordiamo che essa deriva da [BaseEntity];
  • righe 48–54: inizializziamo gli array [limites, coeffr, coeffn] dell'istanza [AdminData];
  • righe 55–56: inizializziamo le altre proprietà di [AdminData] con le costanti di calcolo delle imposte. Abbiamo fatto in modo di dare gli stessi nomi alle proprietà delle classi [AdminData] e [Constantes], il che semplifica il codice;
  • righe 57–58: l'istanza [AdminData] viene memorizzata nel livello [dao] per essere restituita durante le successive chiamate al metodo [get_admindata];
  • riga 60: viene restituito il valore richiesto dal codice chiamante;
  • righe 61–63: gestione degli errori;
  • righe 64–67: il database viene interrogato una sola volta. Possiamo quindi chiudere la sessione [sqlalchemy];

20.2.4. Test del livello [dao]

Nella versione 4 di questa applicazione, abbiamo creato una classe di test per il livello [business]. Più specificamente, essa testava sia il livello [business] che quello [DAO]. Stiamo riutilizzando questo test per verificare che il livello [DAO] funzioni come previsto. Il livello [business], tuttavia, rimane invariato.

Image

Image

Il test [TestDaoMétier] è il seguente:


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()
  • Non torneremo sui 11 test descritti nella sezione |[business] layer test versione 4|;
  • righe 37–66: eseguiremo lo script di test come una normale applicazione piuttosto che come un UnitTest. La riga 66 è quella che attiverà il framework UnitTest. Nei test precedenti, abbiamo utilizzato il metodo [setUp] per configurare l'esecuzione di ciascun test. Ripetevamo la stessa configurazione 11 volte poiché la funzione [setUp] viene eseguita prima di ogni test. Qui, eseguiamo la configurazione una sola volta. Consiste nel definire le variabili globali [business] alla riga 53 e [admindata] alla riga 56, che saranno poi utilizzate dai metodi di [TestDaoBusiness], ad esempio alla riga 12;
  • righe 39–47: lo script di test si aspetta un parametro [mysql / pgres] che indichi se si sta utilizzando un database MySQL o PostgreSQL;
  • righe 50–51: il test viene configurato;
  • riga 53: il livello [business] viene recuperato dalla configurazione;
  • riga 56: facciamo lo stesso con il livello [dao]. Recuperiamo quindi l'istanza [admindata], che incapsula i dati necessari per calcolare l'imposta;
  • I test hanno dimostrato che il metodo [unittest.main()] alla riga 66 non ignorava il parametro [mysql / pgres] passato allo script, ma gli attribuiva invece un significato diverso. La riga 63 garantisce che questo metodo non abbia più alcun parametro;

Creiamo due configurazioni di esecuzione:

Image

Image

Se eseguiamo una di queste due configurazioni, otteniamo i seguenti risultati:


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
  • Righe 5 e 7: tutti gli 11 test superati;

Si noti che questi test verificano solo 11 casi di calcolo delle imposte. Il loro esito positivo può comunque essere sufficiente a darci fiducia nel livello [dao].

20.2.5. Lo script principale

Image

Image

Lo script principale [main] è lo stesso della versione 4:

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

Note

  • righe 1-10: recuperiamo il parametro [mysql / pgres], che specifica il DBMS da utilizzare;
  • righe 12–14: l'applicazione viene configurata;
  • righe 16-17: viene importata la classe [ImpôtsError]. Ne avremo bisogno alla riga 38;
  • righe 19-21: recuperiamo i riferimenti ai livelli dell'applicazione;
  • riga 25: richiediamo i dati dell'amministrazione fiscale dal livello [dao]. Il livello [business] ne ha bisogno per calcolare l'imposta;
  • riga 27: recuperiamo i dati dei contribuenti (id, stato civile, figli, stipendio) in un elenco;
  • righe 29–30: se questo elenco è vuoto, viene generata un'eccezione;
  • righe 32–35: calcoliamo l'imposta per gli elementi nell'elenco [contribuenti];
  • riga 37: si scrivono i risultati nel file JSON [results.json];
  • righe 38–40: gestiamo eventuali errori;

Per eseguire lo script, creiamo due |configurazioni di esecuzione|:

Image

I risultati ottenuti nel file [results.json] sono quelli della versione 4.

Image

20.3. Applicazione 3: Calcolo delle imposte in modalità interattiva

Presentiamo ora l'applicazione che consente il calcolo interattivo delle imposte. Si tratta di un porting dell'Applicazione 2 dalla Versione 4.

Image

Image

  • Lo script [main] avvia il dialogo con l'utente utilizzando il metodo [ui.run] del livello [ui];
  • Il livello [ui]:
    • utilizza il livello [dao] per recuperare i dati necessari al calcolo dell'imposta;
    • chiede all'utente le informazioni relative al contribuente per il quale deve essere calcolata l'imposta;
    • utilizza il livello [business] per eseguire questo calcolo;

Il file [config_layers] istanzia un livello aggiuntivo:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
def configure(config: dict) -> dict:
    #  dao layer instantiation
    from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
    config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)

    #  instantiation layer [business]
    from ImpôtsMétier import ImpôtsMétier
    config['métier'] = ImpôtsMétier()

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

    #  return the config
    return config

La classe [ImpôtsConsole], righe 11–12, è la stessa della |versione 4|.

Lo script principale [main] è il seguente:

 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
#  a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
    sgbd = sys.argv[1].lower()
    erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
    print(f"syntaxe : {syntaxe}")
    sys.exit()

#  configure the application
import config
config = config.configure({'sgbd': sgbd})

#  syspath is configured - imports can be made
from ImpôtsError import ImpôtsError

#  we recover the [ui] layer
ui = config["ui"]

#  code
try:
    #  execution of the [ui] layer
    ui.run()
except ImpôtsError as ex1:
    #  the error message is displayed
    print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
    #  the error message is displayed
    print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
    #  executed in all cases
    print("Travail terminé...")
  • righe 1-10: lo script si aspetta un parametro [mysql / pgres] che specifichi il DBMS da utilizzare;
  • righe 12-14: l'applicazione viene configurata;
  • righe 19-20: il livello [ui] viene recuperato dalla configurazione;
  • riga 25: viene eseguito;

I risultati sono identici a quelli della |versione 4|. Non potrebbe essere altrimenti, dato che tutte le interfacce della versione 4 sono state mantenute nella versione 5.

https://stahe.github.io
Questo tutorial scritto da Serge Tahé è reso disponibile al pubblico secondo i termini della Creative Commons Attribuzione – Non commerciale – Condividi allo stesso modo 3.0 Unported.