20. Esercizio pratico: versione 5
Svilupperemo tre applicazioni:
- l'applicazione 1 inizializzerà il database che sostituirà il file [admindata.json] della versione 4;
- l'applicazione 2 effettuerà il calcolo delle imposte in modalità batch;
- l’applicazione 3 effettuerà il calcolo delle imposte in modalità interattiva;
20.1. Applicazione 1: inizializzazione del database
L’applicazione 1 avrà la seguente struttura:
Si tratta di un’evoluzione dell’architettura della versione 4 (paragrafo |Versione 4|): i dati fiscali saranno contenuti in un database anziché in un file jSON. Il livello [dao] verrà modificato per implementare questa modifica.
20.1.1. Il file [admindata.json]
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
}
Utilizzeremo come colonne del database le chiavi di questo dizionario.
20.1.2. Creazione dei database
Come illustrato nel paragrafo |Creazione di un database MySQL|, creiamo un database MySQL denominato [dbimpots-2019], di proprietà dell’utente [admimpots] con password [mdpimpots]. In [phpMyAdmin] si ottiene quanto segue:
Allo stesso modo, come illustrato nel paragrafo |Creazione di un database PostgreSQL|, creiamo un database PostgreSQL denominato [dbimpots-2019], di proprietà dell’utente [admimpots] con password [mdpimpots]. In [pgAdmin] si ottiene quanto segue:
I database sono stati creati, ma per il momento non contengono alcuna tabella. Queste verranno create da ORM e [sqlalchemy].
20.1.3. Le entità mappate da [sqlalchemy]
Creeremo due tabelle per incapsulare i dati di [admindata.json]:
Definita da [sqlalchemy], la tabella [tbtranches] raccoglierà i dati delle tabelle [limites, coeffr, coeffn] del dizionario [admindata.json]:
# tabella delle fasce d'imposta
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] raccoglierà le costanti del dizionario [admindata.json]:
# la tabella delle costanti
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 con queste due tabelle saranno le seguenti:
L’entità [Constantes] incapsula le costanti del dizionario [admindata.json]:
from BaseEntity import BaseEntity
# classe contenitore dei dati dell'amministrazione fiscale
class Constantes(BaseEntity):
# chiavi escluse dalla dichiarazione della classe
excluded_keys = ["_sa_instance_state"]
# chiavi consentite
@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 [Constantes] 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 ripreso i nomi utilizzati nel dizionario [admindata.json] per facilitare la scrittura del codice;
L’entità [Tranche] incapsula una riga delle tre tabelle [limites, coeffr, coeffn] del dizionario [admindata.json]:
from BaseEntity import BaseEntity
# classe contenitore dei dati dell'amministrazione fiscale
class Tranche(BaseEntity):
# chiavi escluse dallo stato della classe
excluded_keys = ["_sa_instance_state"]
# chiavi consentite
@staticmethod
def get_allowed_keys() -> list:
return ["id", "limite", "coeffr", "coeffn"]
- riga 5: la classe [Tranche] estende la classe [BaseEntity];
- riga 7: dalle proprietà del dizionario [asdict] dell’entità viene esclusa la proprietà [_sa_instance_state] aggiunta da [sqlalchemy];
- righe 10-12: le proprietà della classe;
La mappatura tra le entità [Constantes, Tranche] e le tabelle [constantes, tranches] sarà la seguente:
…
# la tabella delle costanti
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)
)
# tabella delle fasce d'imposta
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)
)
# mappature
from Tranche import Tranche
mapper(Tranche, tranches_table)
from Constantes import Constantes
mapper(Constantes, constantes_table)
- le mappature sono definite alle righe 24-29. In esse si è omesso di stabilire le corrispondenze 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 riportato nelle tabelle i nomi delle proprietà delle entità mappate. Ciò facilita la scrittura del codice e la sua comprensione;
20.1.4. Il file di configurazione di [sqlalchemy]
Abbiamo appena illustrato in dettaglio una parte della configurazione di [sqlalchemy]. Il file [config_database] nella sua interezza è il seguente:
def configure(config: dict) -> dict:
# configurazione SQLAlchemy
from sqlalchemy import create_engine, Table, Column, Integer, MetaData, Float
from sqlalchemy.orm import mapper, sessionmaker
# stringhe di connessione ai database utilizzati
connection_strings = {
'mysql': "mysql+mysqlconnector://admimpots:mdpimpots@localhost/dbimpots-2019",
'pgres': "postgresql+psycopg2://admimpots:mdpimpots@localhost/dbimpots-2019"
}
# stringa di connessione al database in uso
engine = create_engine(connection_strings[config['sgbd']])
# metadati
metadata = MetaData()
# la tabella delle costanti
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)
)
# tabella delle fasce d'imposta
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)
)
# mappature
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)
# una sessione
session = session_factory()
# si registrano alcune informazioni
config['database'] = {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
"constantes_table": constantes_table, "session": session}
# risultato
return config
- riga 1: la funzione [configure] riceve come parametro un dizionario la cui chiave [sgbd] le indica quale SGBD utilizzare: MySQL (mysql) o PostgreSQL (pgres);
- righe 6-12: si seleziona il database richiesto dalla configurazione;
- righe 14-44: mappature entità/tabelle. Queste mappature sono semplici poiché non esiste alcun collegamento tra le tabelle [tranches] e [constantes]. Sono indipendenti. Non vi è quindi alcuna chiave esterna da gestire tra l’una e l’altra;
- righe 46-51: si crea la sessione di lavoro dell’applicazione [session];
- righe 53-58: le informazioni utili vengono inserite nel dizionario di configurazione, che viene poi restituito;
20.1.5. Il livello [dao]
Torniamo all’architettura dell’applicazione 1 da realizzare:
Il livello [dao] [1] deve leggere il file [admindata.json] [2] e trasferirne il contenuto in uno dei database [3, 4];
Il livello [dao] presenta l'interfaccia [1] ed è implementato dalla classe [2].
L'interfaccia [InterfaceDao4TransferAdminData2Database] è la seguente:
# importazioni
from abc import ABC, abstractmethod
# interfaccia InterfaceImpôtsUI
class InterfaceDao4TransferAdminData2Database(ABC):
# trasferimento dei dati fiscali in un database
@abstractmethod
def transfer_admindata_in_database(self:object):
pass
- righe 8-10: l’interfaccia presenta un solo metodo, [transfer_admindata_in_database], senza parametri. Poiché questo metodo richiede dei parametri (quale file?, quale database?), ciò significa che questi saranno passati al costruttore delle classi che implementano tale interfaccia;
La classe [DaoTransferAdminDataFromJsonFile2Database] implementa l’interfaccia [InterfaceDao4TransferAdminData2Database] nel modo seguente:
# importazioni
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):
# produttore
def __init__(self, config: dict):
self.config = config
# trasferimento
def transfer_admindata_in_database(self) -> None:
# inizializzazioni
session = None
config = self.config
try:
# si recuperano i dati dall'amministrazione fiscale
with codecs.open(config["admindataFilename"], "r", "utf8") as fd:
# trasferimento del contenuto in un dizionario
admindata = json.load(fd)
# si recupera la configurazione del database
database = config["database"]
# eliminazione delle due tabelle dal database
# checkfirst=True: verifica innanzitutto che la tabella esista
database["tranches_table"].drop(database["engine"], checkfirst=True)
database["constantes_table"].drop(database["engine"], checkfirst=True)
# ricreazione delle tabelle a partire dalle mappature
database["metadata"].create_all(database["engine"])
# la sessione [sqlalchemy] corrente
session = database["session"]
# si compila la tabella delle fasce d'imposta
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]}))
# si compila la tabella delle costanti
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"]
}))
# convalida della sessione [sqlalchemy]
session.commit()
except (IntegrityError, DatabaseError, InterfaceError) as erreur:
# si rilancia l'eccezione in un'altra forma
raise ImpôtsError(17, f"{erreur}")
finally:
# si liberano le risorse della sessione
if session:
session.close()
- riga 13: la classe [DaoTransferAdminDataFromJsonFile2Database] implementa l’interfaccia [InterfaceDao4TransferAdminData2Database];
- righe 15-17: il costruttore della classe riceve come parametro il dizionario della configurazione. Verranno utilizzate le seguenti chiavi:
- [admindataFilename] (riga 27): il nome del file jSON contenente i dati dell’amministrazione fiscale da trasferire nel database;
- [database] riga 32: la configurazione [sqlalchemy] dell’applicazione;
- righe 34-37: eliminazione delle tabelle [constantes] e [tranches], se presenti;
- righe 39-40: ricreazione delle due tabelle;
- riga 43: si recupera la sessione [sqlalchemy] presente nella configurazione;
- righe 45-51: le tabelle [limites, coeffr, coeffn] del dizionario [admindata] vengono inserite nella sessione. A tal fine, si inseriscono nella sessione le istanze dell’entità [Tranche];
- righe 52-64: un'istanza dell'entità [Constantes] viene inserita nella sessione;
- righe 66-67: la sessione viene convalidata. Se i dati della sessione non erano ancora presenti nel database, vengono inseriti in quel momento;
- righe 68-70: gestione di un eventuale errore;
- righe 71-74: la sessione viene chiusa. Ciò è possibile poiché il livello [dao] viene utilizzato una sola volta;
20.1.6. Configurazione dell’applicazione
L’applicazione è configurata da tre file [1]:
- [config] è il file di configurazione generale. È questo che configura l’applicazione [main]. È coadiuvato dagli altri due file:
- [config_database], che abbiamo già esaminato e che configura ORM e [sqlalchemy];
- [config_layers], che configura i livelli dell’applicazione;
Il file [config] è il seguente:
def configure(config: dict) -> dict:
# [config] ha la chiave [sgbd] che vale:
# [mysql] per gestire un database MySQL
# [pgres] per gestire un database PostgreSQL
import os
# fase 1 ---
# si imposta il Python Path dell'applicazione
# percorso assoluto della cartella contenente questo script
script_dir = os.path.dirname(os.path.abspath(__file__))
# root_dir (da modificare se necessario)
root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
# percorsi assoluti delle dipendenze
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",
# cartelle locali
f"{script_dir}",
f"{script_dir}/../../interfaces",
f"{script_dir}/../../services",
f"{script_dir}/../../entities",
]
# si imposta il syspath
from myutils import set_syspath
set_syspath(absolute_dependencies)
# fase 2 ------
# si completa la configurazione dell'applicazione
config.update({
# percorsi assoluti dei file di dati
"admindataFilename": f"{script_dir}/../../data/input/admindata.json"
})
# fase 3 ------
# configurazione del database
import config_database
config = config_database.configure(config)
# fase 4 ------
# istanziazione dei livelli dell'applicazione
import config_layers
config = config_layers.configure(config)
# si esegue la configurazione
return config
- righe 8-36: si costruisce il Python Path dell’applicazione;
- righe 38-43: si inserisce nella configurazione il percorso del file [admindata.json];
- righe 45-48: configurazione di [sqlalchemy];
- righe 50-53: si istanziano i livelli dell'applicazione;
- riga 56: si restituisce la configurazione generale;
Il file [config_layers] è il seguente:
def configure(config: dict) -> dict:
# istanziazione del livello [dao]
from DaoTransferAdminDataFromJsonFile2Database import DaoTransferAdminDataFromJsonFile2Database
config['dao'] = DaoTransferAdminDataFromJsonFile2Database(config)
# si restituisce la configurazione
return config
- righe 3-4: istanziazione del livello [dao]. Abbiamo visto che il costruttore della classe [DaoTransferAdminDataFromJsonFile2Database] richiede come parametro il dizionario della configurazione generale dell’applicazione;
- riga 4: il riferimento al livello [dao] viene inserito nella configurazione;
- riga 7: si restituisce la configurazione;
20.1.7. Lo script [main] dell’applicazione
Lo script principale [main] è il seguente:
# si attende un parametro mysql o 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()
# si configura l'applicazione
import config
config = config.configure({'sgbd': sgbd})
# il syspath è impostato - è possibile eseguire le importazioni
from ImpôtsError import ImpôtsError
# si recupera il livello [dao]
dao = config["dao"]
# codice
try:
# trasferimento dei dati nel database
dao.transfer_admindata_in_database()
except ImpôtsError as ex1:
# viene visualizzato l'errore
print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
# viene visualizzato l'errore
print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
# fine
print("Terminé...")
- righe 1-10: si attende un parametro. Si verifica che sia presente e corretto;
- righe 12-14: si configura l'applicazione (impostazioni generali, SQLAlchemy, livelli) passando come parametro il tipo di SGBD scelto;
- righe 19-20: sarà necessario il livello [dao]. Lo si recupera;
- riga 25: si esegue il trasferimento nel database. Tutte le informazioni necessarie al metodo [transfer_admindata_in_database] sono disponibili nelle proprietà dello strato [dao] della riga 20. È da lì che le recupererà;
Dopo l’esecuzione con la base MySQL, questa contiene i seguenti elementi (phpMyAdmin):
Nella colonna [3] si vedono i valori assegnati da MySQL alla chiave primaria [id]. La numerazione parte da 1. La schermata sopra riportata è stata ottenuta dopo diverse esecuzioni dello script.
Con il database PostgreSQL i risultati sono i seguenti:
- si fa clic con il tasto destro su [1], quindi su [2-3];
- in [4] sono presenti i dati relativi alle fasce di imposta;
Si ripete la stessa operazione per la tabella delle costanti [tbconstantes]:
20.2. Applicazione 2: calcolo dell’imposta in modalità batch
20.2.1. Architettura
L’applicazione per il calcolo delle imposte della versione 4 utilizzava la seguente architettura:
Il livello [dao] implementa un'interfaccia [InterfaceImpôtsDao]. Abbiamo creato una classe che implementa tale interfaccia:
- [ImpôtsDaoWithAdminDataInJsonFile] che recuperava i dati fiscali da un file jSON. Questa era la versione 3;
Implementeremo l’interfaccia [InterfaceImpôtsDao] tramite una nuova classe [ImpotsDaoWithTaxAdminDataInDatabase] che recupererà i dati dall’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 a rispettare l’interfaccia [InterfaceImpôtsDao], il livello [métier] non dovrà essere modificato.
La nuova architettura sarà la seguente:
20.2.2. Configurazione dell’applicazione
Il file di configurazione [config_database] rimane invariato rispetto all’applicazione 1. La configurazione [config] include nuovi elementi:
# fase 2 ------
# si completa la configurazione dell'applicazione
config.update({
# percorsi assoluti dei file di dati
"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] evolve come segue:
def configure(config: dict) -> dict:
# istanziazione del livello DAO
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# istanziazione del livello [métier]
from ImpôtsMétier import ImpôtsMétier
config['métier'] = ImpôtsMétier()
# si restituisce la configurazione
return config
- righe 3-4: il livello [dao] è ora implementato dalla classe [ImpotsDaoWithAdminDataInDatabase]. Questa classe è nuova ma implementa la stessa interfaccia [InterfaceDao] della versione 4 dell’esercizio applicativo;
- righe 7-8: il livello [métier] è implementato dalla classe [ImpôtsMétier]. Si tratta della classe utilizzata nella versione 4 dell’esercizio pratico;
20.2.3. Il livello [dao]
La classe di implementazione [ImpotsDaoWithAdminDataInDatabase] dell'interfaccia [InterfaceImpôtsDao] sarà la seguente:
# importazioni
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):
# costruttore
def __init__(self, config: dict):
# config["taxPayersFilename"]: il nome del file di testo dei contribuenti
# config["taxPayersResultsFilename"]: il nome del file jSON dei risultati
# config["errorsFilename"]: registra gli errori rilevati in taxPayersFilename
# config["database"]: configurazione del database
# inizializzazione della classe Parent
AbstractImpôtsDao.__init__(self, config)
# memorizzazione dei parametri
self.__config = config
# dati di amministrazione
self.__admindata = None
# implementazione dell'interfaccia
def get_admindata(self):
# admindata è stato memorizzato?
if self.__admindata:
return self.__admindata
# si esegue una richiesta in BD
session = None
config = self.__config
try:
# una sessione
database_config = config["database"]
session = database_config["session"]
# si legge la tabella delle fasce d'imposta
tranches = session.query(Tranche).all()
# si legge la tabella delle costanti (una sola riga)
constantes = session.query(Constantes).first()
# si crea l'istanza admindata
admindata = AdminData()
# si creano le tabelle dei limiti, 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))
# si aggiungono le costanti
admindata.fromdict(constantes.asdict())
# si salva «admindata»
self.__admindata = admindata
# si restituisce il valore
return self.__admindata
except (IntegrityError, DatabaseError, InterfaceError) as erreur:
# si rilancia l'eccezione in un'altra forma
raise ImpôtsError(27, f"{erreur}")
finally:
# si chiude la sessione
if session:
session.close()
Note
- riga 11: la classe [ImpotsDaoWithAdminDataInDatabase] eredita dalla classe [AbstractImpôtsDao] presente nella versione 4. È noto che quest'ultima implementa l'interfaccia [InterfaceDao] presente nella stessa versione. È proprio il rispetto di questa interfaccia che ci permette di non modificare il livello [métier];
- riga 13: il costruttore della classe riceve come parametro il dizionario della 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 [résultats.json];
- [get_admindata] non è implementato;
- riga 22: si memorizza la configurazione passata come parametri;
- 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 tale 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 volte successive, viene restituito l’oggetto [self.__admindata];
- righe 36-37: si recupera la sessione [sqlalchemy] che è stata creata durante la configurazione dell’applicazione tramite [config_database];
- riga 40: si recuperano le fasce di imposta in un elenco;
- riga 43: si recuperano le costanti per il calcolo dell’imposta;
- riga 46: si crea un'istanza della classe [AdminData]. Si ricorda che essa deriva da [BaseEntity];
- righe 48-54: si inizializzano gli array [limites, coeffr, coeffn] dell’istanza [AdminData];
- righe 55-56: si inizializzano le altre proprietà di [AdminData] con le costanti del calcolo dell’imposta. Si è avuto cura di assegnare 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 restituirla nelle successive chiamate al metodo [get_admindata];
- riga 60: si restituisce il valore richiesto dal codice chiamante;
- righe 61-63: gestione di un eventuale errore;
- righe 64-67: il database è oggetto di una sola query. È quindi possibile chiudere la sessione [sqlalchemy];
20.2.4. Test del livello [dao]
Nella versione 4 di questa applicazione, avevamo creato una classe di test per il livello [métier]. Più precisamente, essa testava contemporaneamente i livelli [métier] e [dao]. Riprendiamo questo test per verificare che il livello [dao] funzioni come previsto. Infatti, il livello [métier] rimane invariato.
Il test [TestDaoMétier] è il seguente:
import unittest
class TestDaoMétier(unittest.TestCase):
def test_1(self) -> None:
from TaxPayer import TaxPayer
# {'sposato': 'sì', 'figli': 2, 'stipendio': 55555,
# 'imposta': 2814, 'maggiorazione': 0, 'riduzione': 0, 'sconto': 0, 'aliquota': 0,14}
taxpayer = TaxPayer().fromdict({"marié": "oui", "enfants": 2, "salaire": 55555})
métier.calculate_tax(taxpayer, admindata)
# verifica
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
# {'coniugato': 'sì', 'figli': 3, 'stipendio': 200000,
# 'imposta': 42842, 'maggiorazione': 17283, 'sconto': 0, 'riduzione': 0, 'aliquota': 0,41}
taxpayer = TaxPayer().fromdict({'marié': 'oui', 'enfants': 3, 'salaire': 200000})
métier.calculate_tax(taxpayer, admindata)
# verifiche
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__':
# si attende un parametro mysql o 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()
# si sta configurando l'applicazione
import config
config = config.configure({'sgbd': sgbd})
# livello business
métier = config['métier']
try:
# gestione dati
admindata = config['dao'].get_admindata()
except BaseException as ex:
# visualizzazione
print((f"L'erreur suivante s'est produite : {ex}"))
# fine
sys.exit()
# si invia il parametro ricevuto allo script
sys.argv.pop()
# si eseguono i metodi di test
print("tests en cours...")
unittest.main()
- Non torniamo sui 11 test descritti nel paragrafo |test livello [métier] versione 4|;
- righe 37-66: eseguiremo lo script dei test come una normale applicazione e non come un test UnitTest. È la riga 66 che attiverà il framework UnitTest. Nei test precedenti, utilizzavamo il metodo [setUp] per configurare l’esecuzione di ciascun test. Ripetevamo 11 volte la stessa configurazione poiché la funzione [setUp] viene eseguita prima di ogni test. In questo caso, la configurazione viene eseguita una sola volta. Consiste nel definire le variabili globali [métier] alla riga 53 e [admindata] alla riga 56, che saranno poi utilizzate dai metodi di [TestDaoMétier], ad esempio alla riga 12;
- righe 39-47: lo script di test attende un parametro [mysql / pgres] che indica se si utilizza un database MySQL o PostgreSQL;
- righe 50-51: il test viene configurato;
- riga 53: si recupera il livello [métier] dalla configurazione;
- riga 56: si procede allo stesso modo con il livello [dao]. Si recupera quindi l’istanza [admindata] che incapsula i dati necessari al calcolo dell’imposta;
- i test hanno dimostrato che il metodo [unittest.main()] della riga 66 non ignorava il parametro [mysql / pgres] ricevuto dallo script, ma gli attribuiva un significato diverso. La riga 63 fa in modo che questo metodo non abbia più alcun parametro;
Creiamo due configurazioni di esecuzione:
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: gli 11 test sono stati superati;
Ricordiamo che questi test verificano solo 11 casi di calcolo dell’imposta. Il loro esito positivo può tuttavia essere sufficiente per darci fiducia nel livello [dao].
20.2.5. Lo script principale
Lo script principale [main] è lo stesso della versione 4:
# si attende un parametro mysql o 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()
# si configura l'applicazione
import config
config = config.configure({'sgbd': sgbd})
# il syspath è impostato - è possibile eseguire le importazioni
from ImpôtsError import ImpôtsError
# si recuperano i livelli dell'applicazione (sono già istanziati)
dao = config["dao"]
métier = config["métier"]
try:
# recupero delle fasce d'imposta
admindata = dao.get_admindata()
# lettura dei dati dei contribuenti
taxpayers = dao.get_taxpayers_data()["taxpayers"]
# dei contribuenti?
if not taxpayers:
raise ImpôtsError(57, f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
# calcolo dell'imposta dei contribuenti
for taxPayer in taxpayers:
# taxPayer è sia un parametro di input che di output
# taxPayer verrà modificato
métier.calculate_tax(taxPayer, admindata)
# scrittura dei risultati in un file di testo
dao.write_taxpayers_results(taxpayers)
except ImpôtsError as erreur:
# visualizzazione dell'errore
print(f"L'erreur suivante s'est produite : {erreur}")
finally:
# terminato
print("Travail terminé...")
Note
- righe 1-10: si recupera il parametro [mysql / pgres] che indica il SGBD 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: si recuperano i riferimenti ai livelli dell’applicazione;
- riga 25: si richiedono al livello [dao] i dati dell’amministrazione fiscale. Il livello [métier] ne ha bisogno per il calcolo dell’imposta;
- riga 27: si recuperano in un elenco i dati (ID, stato civile, figli, stipendio) dei contribuenti;
- righe 29-30: se l’elenco è vuoto, viene generata un’eccezione;
- righe 32-35: calcolo dell’imposta sugli elementi dell’elenco [taxpayers];
- riga 37: scrittura dei risultati nel file jSON[résultats.json];
- righe 38-40: gestione di eventuali errori;
Per l'esecuzione dello script, si creano due |configurazioni di esecuzione|:
I risultati ottenuti nel file [résultats.json] sono quelli della versione 4.
20.3. Applicazione 3: calcolo dell’imposta in modalità interattiva
Presentiamo ora l’applicazione che consente di calcolare l’imposta in modo interattivo. Si tratta di un porting dell’applicazione 2 della versione 4.
- lo script [main] avvia la finestra di dialogo con l’utente tramite il metodo [ui.run] del livello [ui];
- il livello [ui]:
- utilizza il livello [dao] per ottenere i dati necessari al calcolo dell’imposta;
- richiede all'utente le informazioni relative al contribuente per il quale si desidera calcolare l'imposta;
- utilizza il livello [métier] per effettuare tale calcolo;
Il file [config_layers] istanzia un livello aggiuntivo:
def configure(config: dict) -> dict:
# istanziazione del livello DAO
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# istanziazione del livello [métier]
from ImpôtsMétier import ImpôtsMétier
config['métier'] = ImpôtsMétier()
# interfaccia utente
from ImpôtsConsole import ImpôtsConsole
config['ui'] = ImpôtsConsole(config)
# si restituisce la configurazione
return config
La classe [ImpôtsConsole], righe 11-12, è la stessa della |versione 4|.
Lo script principale [main] è il seguente:
# si attende un parametro mysql o 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()
# si configura l'applicazione
import config
config = config.configure({'sgbd': sgbd})
# il syspath è configurato - si possono eseguire le importazioni
from ImpôtsError import ImpôtsError
# si recupera il livello [ui]
ui = config["ui"]
# codice
try:
# esecuzione del livello [ui]
ui.run()
except ImpôtsError as ex1:
# viene visualizzato il messaggio di errore
print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
# viene visualizzato il messaggio di errore
print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
# eseguito in ogni caso
print("Travail terminé...")
- righe 1-10: lo script attende un parametro [mysql / pgres] che indichi il SGBD da utilizzare;
- righe 12-14: l’applicazione viene configurata;
- righe 19-20: si recupera il livello [ui] dalla configurazione;
- riga 25: lo si esegue;
I risultati sono identici a quelli della |versione 4|. Non poteva essere altrimenti, poiché tutte le interfacce della versione 4 sono state mantenute nella versione 5.