20. Anwendungsübung: Version 5
Wir werden drei Anwendungen entwickeln:
- Anwendung 1 initialisiert die Datenbank, die die Datei [admindata.json] aus Version 4 ersetzen wird;
- Anwendung 2 führt die Steuerberechnung im Batch-Modus durch;
- Anwendung 3 führt die Steuerberechnung im interaktiven Modus durch;
20.1. Anwendung 1: Initialisierung der Datenbank
Anwendung 1 wird folgende Architektur aufweisen:
Dies ist eine Weiterentwicklung der Architektur der Version 4 (Abschnitt |Version 4|): Die Steuerdaten werden in einer Datenbank gespeichert, anstatt in einer Datei mit dem Namen jSON. Die Schicht [dao] wird angepasst, um diese Änderung umzusetzen.
20.1.1. Die Datei [admindata.json]
Die Datei [admindata.json] entspricht der Version 4:
{
"limites": [9964, 27519, 73779, 156244, 0],
"coeffr": [0, 0.14, 0.3, 0.41, 0.45],
"coeffn": [0, 1394.96, 5798, 13913.69, 20163.45],
"plafond_qf_demi_part": 1551,
"plafond_revenus_celibataire_pour_reduction": 21037,
"plafond_revenus_couple_pour_reduction": 42074,
"valeur_reduc_demi_part": 3797,
"plafond_decote_celibataire": 1196,
"plafond_decote_couple": 1970,
"plafond_impot_couple_pour_decote": 2627,
"plafond_impot_celibataire_pour_decote": 1595,
"abattement_dixpourcent_max": 12502,
"abattement_dixpourcent_min": 437
}
Wir werden die Schlüssel dieses Wörterbuchs als Spalten der Datenbank verwenden.
20.1.2. Erstellung der Datenbanken
Wie im Abschnitt |Erstellung einer Datenbank MySQL| gezeigt wurde, legen wir eine Datenbank MySQL mit dem Namen [dbimpots-2019] an, die dem Benutzer [admimpots] mit dem Passwort [mdpimpots] gehört. In [phpMyAdmin] ergibt sich daraus Folgendes:
Ebenso wie im Abschnitt |Erstellung einer Datenbank PostgreSQL| gezeigt, legen wir eine Datenbank mit dem Namen PostgreSQL an, die den Namen [dbimpots-2019] trägt und dem Benutzer [admimpots] mit dem Passwort [mdpimpots] gehört. In [pgAdmin] ergibt sich daraus Folgendes:
Die Datenbanken wurden angelegt, enthalten jedoch vorerst keine Tabellen. Diese werden von ORM und [sqlalchemy] erstellt.
20.1.3. Die von [sqlalchemy] zugeordneten Entitäten
Wir werden zwei Tabellen anlegen, um die Daten von [admindata.json] zu kapseln:
Die von [sqlalchemy] definierte Tabelle [tbtranches] wird die Daten der Arrays [limites, coeffr, coeffn] aus dem Wörterbuch [admindata.json] zusammenfassen:
# die Steuertabelle
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)
)
Die durch [sqlalchemy] definierte Tabelle [tbconstantes] fasst die Konstanten des Wörterbuchs [admindata.json] zusammen:
# die Konstantentabelle
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)
)
Die folgenden Entitäten werden diesen beiden Tabellen zugeordnet:
Die Entität [Constantes] fasst die Konstanten des Wörterbuchs [admindata.json] zusammen:
from BaseEntity import BaseEntity
# Containerklasse für Daten der Steuerverwaltung
class Constantes(BaseEntity):
# aus der Klassenbeschreibung ausgeschlossene Schlüssel
excluded_keys = ["_sa_instance_state"]
# Zulässige Schlüssel
@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"]
- Zeile 5: Die Klasse [Constantes] erweitert die Klasse [BaseEntity];
- Zeile 7: Durch das Mapping [sqlalchemy] erhält die Klasse [Constante] die Eigenschaft [_sa_instance_state]. Wir schließen sie aus dem Wörterbuch [asdict] der Entität aus;
- Zeilen 11–23: Die Eigenschaften der Entität. Wir haben die im Wörterbuch [admindata.json] verwendeten Namen übernommen, um das Schreiben des Codes zu vereinfachen;
Die Entität [Tranche] kapselt eine Zeile der drei Tabellen [limites, coeffr, coeffn] aus dem Wörterbuch [admindata.json]:
from BaseEntity import BaseEntity
# Containerklasse für Daten der Steuerverwaltung
class Tranche(BaseEntity):
# aus dem Klassenstatus ausgeschlossene Schlüssel
excluded_keys = ["_sa_instance_state"]
# zulässige Schlüssel
@staticmethod
def get_allowed_keys() -> list:
return ["id", "limite", "coeffr", "coeffn"]
- Zeile 5: Die Klasse [Tranche] erweitert die Klasse [BaseEntity];
- Zeile 7: Die durch [sqlalchemy] hinzugefügte Eigenschaft [_sa_instance_state] wird aus den Eigenschaften des Wörterbuchs [asdict] der Entität ausgeschlossen;
- Zeilen 10–12: die Eigenschaften der Klasse;
Die Zuordnung zwischen den Entitäten [Constantes, Tranche] und den Tabellen [constantes, tranches] sieht wie folgt aus:
…
# die Konstantentabelle
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)
)
# die Steuertabelle
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)
)
# die Mappings
from Tranche import Tranche
mapper(Tranche, tranches_table)
from Constantes import Constantes
mapper(Constantes, constantes_table)
- Die Zuordnungen erfolgen in den Zeilen 24–29. Dabei wurde darauf verzichtet, die Eigenschaften der zugeordneten Entitäten den Tabellen der Datenbank zuzuordnen. Dies ist möglich, wenn die Spaltennamen der Tabellen mit den Namen der Eigenschaften übereinstimmen, denen sie zugeordnet werden sollen. Aus diesem Grund haben wir die Namen der Eigenschaften der zugeordneten Entitäten in die Tabellen übernommen. Dies erleichtert das Schreiben und das Verständnis des Codes;
20.1.4. Die Konfigurationsdatei von [sqlalchemy]
Wir haben soeben einen Teil der Konfiguration von [sqlalchemy] erläutert. Die vollständige Datei [config_database] lautet wie folgt:
def configure(config: dict) -> dict:
# SQLAlchemy-Konfiguration
from sqlalchemy import create_engine, Table, Column, Integer, MetaData, Float
from sqlalchemy.orm import mapper, sessionmaker
# Verbindungsstrings zu den verwendeten Datenbanken
connection_strings = {
'mysql': "mysql+mysqlconnector://admimpots:mdpimpots@localhost/dbimpots-2019",
'pgres': "postgresql+psycopg2://admimpots:mdpimpots@localhost/dbimpots-2019"
}
# Verbindungszeichenfolge zur verwendeten Datenbank
engine = create_engine(connection_strings[config['sgbd']])
# Metadaten
metadata = MetaData()
# die Konstantentabelle
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)
)
# Steuerklassen-Tabelle
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)
# die Session-Factory
session_factory = sessionmaker()
session_factory.configure(bind=engine)
# eine Sitzung
session = session_factory()
# Es werden bestimmte Informationen gespeichert
config['database'] = {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
"constantes_table": constantes_table, "session": session}
# Ergebnis
return config
- Zeile 1: Die Funktion [configure] erhält als Parameter ein Wörterbuch, dessen Schlüssel [sgbd] angibt, welches SGBD verwendet werden soll: MySQL (MySQL) oder PostgreSQL (PostgreSQL);
- Zeilen 6–12: Hier wird die in der Konfiguration angegebene Datenbank ausgewählt;
- Zeilen 14–44: Zuordnungen von Entitäten zu Tabellen. Diese Zuordnungen sind einfach, da zwischen den Tabellen [tranches] und [constantes] keine Verknüpfungen bestehen. Sie sind voneinander unabhängig. Es müssen daher keine Fremdschlüssel zwischen den beiden Tabellen verwaltet werden;
- Zeilen 46–51: Es wird die Arbeitssitzung [session] der Anwendung angelegt;
- Zeilen 53–58: Die relevanten Informationen werden in das Konfigurationswörterbuch aufgenommen und dieses wird zurückgegeben;
20.1.5. Die Schicht [dao]
Kehren wir zur Architektur der zu erstellenden Anwendung 1 zurück:
Die Schicht [dao] [1] muss die Datei [admindata.json] [2] lesen und deren Inhalt in eine der Datenbanken [3, 4] übertragen;
Die Schicht [dao] verfügt über die Schnittstelle [1] und wird durch die Klasse [2] implementiert.
Die Schnittstelle [InterfaceDao4TransferAdminData2Database] lautet wie folgt:
# Importe
from abc import ABC, abstractmethod
# Schnittstelle InterfaceImpôtsUI
class InterfaceDao4TransferAdminData2Database(ABC):
# Übertragung der Steuerdaten in eine Datenbank
@abstractmethod
def transfer_admindata_in_database(self:object):
pass
- Zeilen 8–10: Die Schnittstelle enthält nur eine Methode [transfer_admindata_in_database] ohne Parameter. Da diese Methode Parameter benötigt (welche Datei?, welche Datenbank?), bedeutet dies, dass diese an den Konstruktor der Klassen übergeben werden, die diese Schnittstelle implementieren;
Die Klasse [DaoTransferAdminDataFromJsonFile2Database] implementiert die Schnittstelle [InterfaceDao4TransferAdminData2Database] wie folgt:
# Importe
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):
# Hersteller
def __init__(self, config: dict):
self.config = config
# Übertragung
def transfer_admindata_in_database(self) -> None:
# Initialisierungen
session = None
config = self.config
try:
# Daten von der Steuerbehörde abrufen
with codecs.open(config["admindataFilename"], "r", "utf8") as fd:
# Übertragung des Inhalts in ein Wörterbuch
admindata = json.load(fd)
# Abfrage der Datenbankkonfiguration
database = config["database"]
# Löschen der beiden Tabellen aus der Datenbank
# checkfirst=True: Prüft zunächst, ob die Tabelle vorhanden ist
database["tranches_table"].drop(database["engine"], checkfirst=True)
database["constantes_table"].drop(database["engine"], checkfirst=True)
# Neuanlegen der Tabellen anhand der Zuordnungen
database["metadata"].create_all(database["engine"])
# die aktuelle Sitzung [sqlalchemy]
session = database["session"]
# Die Tabelle der Steuerklassen wird gefüllt
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]}))
# Die Konstantentabelle wird gefüllt
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"]
}))
# Sitzungsvalidierung [sqlalchemy]
session.commit()
except (IntegrityError, DatabaseError, InterfaceError) as erreur:
# Die Ausnahme wird in einer anderen Form erneut ausgelöst
raise ImpôtsError(17, f"{erreur}")
finally:
# Die Ressourcen der Sitzung werden freigegeben
if session:
session.close()
- Zeile 13: Die Klasse [DaoTransferAdminDataFromJsonFile2Database] implementiert die Schnittstelle [InterfaceDao4TransferAdminData2Database];
- Zeilen 15–17: Der Konstruktor der Klasse erhält das Konfigurationswörterbuch als Parameter. Die folgenden Schlüssel werden verwendet:
- [admindataFilename] (Zeile 27): der Name der Datei jSON, die die Daten der Steuerbehörde enthält, die in die Datenbank übertragen werden sollen;
- [database], Zeile 32: die Konfiguration [sqlalchemy] der Anwendung;
- Zeilen 34–37: Löschen der Tabellen [constantes] und [tranches], sofern vorhanden;
- Zeilen 39–40: Neuerstellung der beiden Tabellen;
- Zeile 43: Die in der Konfiguration vorhandene Sitzung [sqlalchemy] wird abgerufen;
- Zeilen 45–51: Die Tabellen [limites, coeffr, coeffn] aus dem Wörterbuch [admindata] werden in die Sitzung aufgenommen. Dazu werden Instanzen der Entität [Tranche] in die Sitzung aufgenommen;
- Zeilen 52–64: Eine Instanz der Entität [Constantes] wird in die Sitzung übernommen;
- Zeilen 66–67: Die Sitzung wird validiert. Falls die Daten der Sitzung noch nicht in der Datenbank vorhanden waren, werden sie zu diesem Zeitpunkt dort gespeichert;
- Zeilen 68–70: Behandlung eines möglichen Fehlers;
- Zeilen 71–74: Die Sitzung wird geschlossen. Dies ist möglich, da die Schicht [dao] nur einmal verwendet wird;
20.1.6. Konfiguration der Anwendung
Die Anwendung wird über drei Dateien konfiguriert: [1]:
- [config] ist die allgemeine Konfigurationsdatei. Sie konfiguriert die Anwendung [main]. Dabei wird sie von den beiden anderen Dateien unterstützt:
- [config_database], die wir bereits behandelt haben und die die Dateien ORM und [sqlalchemy] konfiguriert;
- [config_layers], die die Schichten der Anwendung konfiguriert;
Die Datei [config] sieht wie folgt aus:
def configure(config: dict) -> dict:
# [config] hat den Schlüssel [sgbd] mit folgendem Wert:
# [mysql] zur Verwaltung einer Datenbank MySQL
# [pgres] zur Verwaltung einer Datenbank mit dem Namen PostgreSQL
import os
# Schritt 1 ---
# Wir legen den Python-Pfad der Anwendung fest
# Absoluter Pfad zum Ordner dieses Skripts
script_dir = os.path.dirname(os.path.abspath(__file__))
# root_dir (ggf. ändern)
root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
# Absolute Pfade der Abhängigkeiten
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",
# lokale Ordner
f"{script_dir}",
f"{script_dir}/../../interfaces",
f"{script_dir}/../../services",
f"{script_dir}/../../entities",
]
# Der Syspath wird festgelegt
from myutils import set_syspath
set_syspath(absolute_dependencies)
# Schritt 2 ------
# Die Konfiguration der Anwendung wird vervollständigt
config.update({
# Absolute Pfade der Datendateien
"admindataFilename": f"{script_dir}/../../data/input/admindata.json"
})
# Schritt 3 ------
# Datenbankkonfiguration
import config_database
config = config_database.configure(config)
# Schritt 4 ------
# Instanziierung der Anwendungsschichten
import config_layers
config = config_layers.configure(config)
# Konfiguration wird übernommen
return config
- Zeilen 8–36: Der Python-Pfad der Anwendung wird erstellt;
- Zeilen 38–43: Der Pfad zur Datei „[admindata.json]“ wird in die Konfiguration aufgenommen;
- Zeilen 45–48: Konfiguration von [sqlalchemy];
- Zeilen 50–53: Instanziierung der Anwendungsschichten;
- Zeile 56: Die allgemeine Konfiguration wird zurückgegeben;
Die Datei [config_layers] sieht wie folgt aus:
def configure(config: dict) -> dict:
# Instanziierung der Schicht [dao]
from DaoTransferAdminDataFromJsonFile2Database import DaoTransferAdminDataFromJsonFile2Database
config['dao'] = DaoTransferAdminDataFromJsonFile2Database(config)
# Konfiguration wird zurückgegeben
return config
- Zeilen 3–4: Instanziierung der Schicht [dao]. Wir haben gesehen, dass der Konstruktor der Klasse [DaoTransferAdminDataFromJsonFile2Database] als Parameter das Wörterbuch der allgemeinen Anwendungskonfiguration erwartet;
- Zeile 4: Die Referenz auf die Schicht [dao] wird in die Konfiguration aufgenommen;
- Zeile 7: Die Konfiguration wird zurückgegeben;
20.1.7. Das Skript [main] der Anwendung
Das Hauptskript [main] lautet wie folgt:
# Es wird ein MySQL- oder PostgreSQL-Parameter erwartet
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()
# Die Anwendung wird konfiguriert
import config
config = config.configure({'sgbd': sgbd})
# Der Syspath ist eingerichtet – die Importe können durchgeführt werden
from ImpôtsError import ImpôtsError
# die Ebene wird abgerufen [dao]
dao = config["dao"]
# Code
try:
# Datenübertragung in die Datenbank
dao.transfer_admindata_in_database()
except ImpôtsError as ex1:
# Fehler wird angezeigt
print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
# Fehler wird angezeigt
print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
# Ende
print("Terminé...")
- Zeilen 1–10: Es wird ein Parameter erwartet. Es wird überprüft, ob dieser vorhanden und korrekt ist;
- Zeilen 12–14: Die Anwendung wird konfiguriert (allgemein, SQLAlchemy, Schichten), indem der ausgewählte Typ von SGBD als Parameter übergeben wird;
- Zeilen 19–20: Wir benötigen die Schicht [dao]. Diese wird abgerufen;
- Zeile 25: Die Übertragung in die Datenbank erfolgt. Alle für die Methode [transfer_admindata_in_database] erforderlichen Informationen sind in den Eigenschaften der Schicht [dao] aus Zeile 20 verfügbar. Dort werden sie abgerufen;
Nach der Ausführung mit der Basis MySQL enthält diese die folgenden Elemente (phpMyAdmin):
In der Spalte [3] sind die Werte zu sehen, die von MySQL dem Primärschlüssel [id] zugewiesen wurden. Die Nummerierung beginnt bei 1. Der obige Screenshot wurde nach mehreren Ausführungen des Skripts erstellt.
Mit der Datenbank PostgreSQL ergeben sich folgende Ergebnisse:
- Man klickt mit der rechten Maustaste auf [1] und anschließend auf [2-3];
- in [4] sind die Daten zu den Steuerklassen tatsächlich vorhanden;
Man wiederholt den Vorgang für die Konstantentabelle [tbconstantes]:
20.2. Anwendung 2: Steuerberechnung im Batch-Modus
20.2.1. Architektur
Die Steuerberechnungsanwendung der Version 4 verwendete die folgende Architektur:
Die Schicht [dao] implementiert eine Schnittstelle [InterfaceImpôtsDao]. Wir haben eine Klasse erstellt, die diese Schnittstelle implementiert:
- [ImpôtsDaoWithAdminDataInJsonFile], die die Steuerdaten aus einer Datei jSON abrief. Das war Version 3;
Wir werden die Schnittstelle [InterfaceImpôtsDao] durch eine neue Klasse [ImpotsDaoWithTaxAdminDataInDatabase] implementieren, die die Daten der Steuerbehörde aus einer Datenbank abruft. Die Schicht [dao] wird wie bisher die Ergebnisse in eine Datei jSON schreiben und die Daten der Steuerzahler aus einer Textdatei abrufen. Wir wissen, dass die Schicht [InterfaceImpôtsDao] nicht geändert werden muss, wenn wir weiterhin die Schnittstelle [InterfaceImpôtsDao] einhalten.
Die neue Architektur sieht wie folgt aus:
20.2.2. Anwendungskonfiguration
Die Konfigurationsdatei [config_database] bleibt unverändert gegenüber der Anwendung 1. Die Konfiguration [config] enthält neue Elemente:
# Schritt 2 ------
# Die Konfiguration der Anwendung wird vervollständigt
config.update({
# Absolute Pfade der Datendateien
"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"
})
- Zeilen 6–8: die absoluten Pfade der von Anwendung 2 verwendeten Textdateien;
Die Konfiguration der Ebenen [config_layers] ändert sich wie folgt:
def configure(config: dict) -> dict:
# Instanziierung der DAO-Schicht
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# Instanziierung der [métier]-Schicht
from ImpôtsMétier import ImpôtsMétier
config['métier'] = ImpôtsMétier()
# Die Konfiguration wird zurückgegeben
return config
- Zeilen 3–4: Die Schicht [dao] wird nun durch die Klasse [ImpotsDaoWithAdminDataInDatabase] implementiert. Diese Klasse ist neu, implementiert jedoch dieselbe Schnittstelle [InterfaceDao] wie Version 4 der Anwendungsübung;
- Zeilen 7–8: Die Schicht [métier] wird durch die Klasse [ImpôtsMétier] implementiert. Dies ist die Klasse, die in Version 4 der Anwendungsübung verwendet wird;
20.2.3. Die Schicht [dao]
Die Implementierungsklasse [ImpotsDaoWithAdminDataInDatabase] der Schnittstelle [InterfaceImpôtsDao] sieht wie folgt aus:
# Importe
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):
# Konstruktor
def __init__(self, config: dict):
# Konfiguration["taxPayersFilename"]: Name der Textdatei mit den Steuerpflichtigen
# config["taxPayersResultsFilename"]: Name der Ergebnisdatei jSON
# config["errorsFilename"]: Speichert die in taxPayersFilename gefundenen Fehler
# config["database"]: Konfiguration der Datenbank
# Initialisierung der Klasse „Parent“
AbstractImpôtsDao.__init__(self, config)
# Speicherung von Parametern
self.__config = config
# Admindaten
self.__admindata = None
# Implementierung der Schnittstelle
def get_admindata(self):
# Wurde „admindata“ gespeichert?
if self.__admindata:
return self.__admindata
# Es wird eine Abfrage in BD durchgeführt
session = None
config = self.__config
try:
# eine Sitzung
database_config = config["database"]
session = database_config["session"]
# Die Tabelle der Steuerklassen wird gelesen
tranches = session.query(Tranche).all()
# Die Konstantentabelle wird gelesen (nur eine Zeile)
constantes = session.query(Constantes).first()
# die Instanz „admindata“ wird angelegt
admindata = AdminData()
# darin werden die Grenzwerttabellen coeffR und coeffN angelegt
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))
# dort werden die Konstanten hinzugefügt
admindata.fromdict(constantes.asdict())
# admindata wird gespeichert
self.__admindata = admindata
# man gibt den Wert zurück
return self.__admindata
except (IntegrityError, DatabaseError, InterfaceError) as erreur:
# die Ausnahme wird in einer anderen Form erneut ausgelöst
raise ImpôtsError(27, f"{erreur}")
finally:
# die Sitzung wird beendet
if session:
session.close()
Hinweise
- Zeile 11: Die Klasse [ImpotsDaoWithAdminDataInDatabase] erbt von der in Version 4 vorgestellten Klasse [AbstractImpôtsDao]. Es ist bekannt, dass letztere die in derselben Version vorgestellte Schnittstelle [InterfaceDao] implementiert. Da diese Schnittstelle eingehalten wird, müssen wir die Schicht [métier] nicht ändern;
- Zeile 13: Der Konstruktor der Klasse erhält als Parameter das Wörterbuch mit der Anwendungskonfiguration;
- Zeile 20: Die übergeordnete Klasse [] wird initialisiert. Sie implementiert teilweise die Schnittstelle [InterfaceDao]:
- [get_taxpayers_data] liest die Datei [taxpayersdata.txt], die die Daten der Steuerzahler enthält;
- [write_taxpayers_results] schreibt die Ergebnisse in die Datei jSON [résultats.json];
- [get_admindata] ist nicht implementiert;
- Zeile 22: Die als Parameter übergebenen Konfigurationsdaten werden gespeichert;
- Zeile 27: Implementierung der Methode [get_admindata] der Schnittstelle [InterfaceDao]:
- Zeilen 28–30: Die Methode [get_admindata] ruft die Daten der Steuerbehörde in einem Objekt vom Typ [AdminData] ab und speichert dieses Objekt in [self.__admindata]. Wird die Methode [get_admindata] mehrmals aufgerufen, wird die Datenbank nicht mehrfach abgefragt. Sie wird nur beim ersten Mal abgefragt. Bei den folgenden Aufrufen wird das Objekt [self.__admindata] zurückgegeben;
- Zeilen 36–37: Die Sitzung [sqlalchemy] wird abgerufen, die bei der Konfiguration der Anwendung durch [config_database] angelegt wurde;
- Zeile 40: Die Steuerklassen werden in einer Liste abgerufen;
- Zeile 43: Die Konstanten für die Steuerberechnung werden abgerufen;
- Zeile 46: Es wird eine Instanz der Klasse [AdminData] erstellt. Zur Erinnerung: Diese Klasse leitet sich von [BaseEntity] ab;
- Zeilen 48–54: Die Arrays [limites, coeffr, coeffn] der Instanz [AdminData] werden initialisiert;
- Zeilen 55–56: Die übrigen Eigenschaften von [AdminData] werden mit den Konstanten der Steuerberechnung initialisiert. Wir haben darauf geachtet, den Eigenschaften der Klassen [AdminData] und [Constantes] dieselben Namen zu geben, was den Code vereinfacht;
- Zeilen 57–58: Die Instanz [AdminData] wird in der Schicht [dao] gespeichert, um sie bei den nächsten Aufrufen der Methode [get_admindata] zurückzugeben;
- Zeile 60: Der vom aufrufenden Code angeforderte Wert wird zurückgegeben;
- Zeilen 61–63: Behandlung eines möglichen Fehlers;
- Zeilen 64–67: Es wird nur eine einzige Abfrage an die Datenbank gesendet. Daher kann die Sitzung [sqlalchemy] geschlossen werden;
20.2.4. Test der Schicht [dao]
In Version 4 dieser Anwendung hatten wir eine Testklasse für die Schicht [métier] erstellt. Genauer gesagt testete diese sowohl die Schichten [métier] als auch [dao]. Wir greifen diesen Test wieder auf, um zu überprüfen, ob die Schicht [dao] wie erwartet funktioniert. Die Schicht [métier] bleibt nämlich unverändert.
Der Test [TestDaoMétier] sieht wie folgt aus:
import unittest
class TestDaoMétier(unittest.TestCase):
def test_1(self) -> None:
from TaxPayer import TaxPayer
# {'verheiratet': 'ja', 'Kinder': 2, 'Gehalt': 55555,
# 'Steuer': 2814, 'Zuschlag': 0, 'Abschlag': 0, 'Ermäßigung': 0, 'Satz': 0,14}
taxpayer = TaxPayer().fromdict({"marié": "oui", "enfants": 2, "salaire": 55555})
métier.calculate_tax(taxpayer, admindata)
# Überprüfung
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
# {'verheiratet': 'ja', 'Kinder': 3, 'Gehalt': 200000,
# 'Steuer': 42842, 'Zuschlag': 17283, 'Abschlag': 0, 'Ermäßigung': 0, 'Satz': 0,41}
taxpayer = TaxPayer().fromdict({'marié': 'oui', 'enfants': 3, 'salaire': 200000})
métier.calculate_tax(taxpayer, admindata)
# Prüfungen
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__':
# Es wird ein MySQL- oder PostgreSQL-Parameter erwartet
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()
# Die Anwendung wird konfiguriert
import config
config = config.configure({'sgbd': sgbd})
# Geschäftslogik
métier = config['métier']
try:
# Admindaten
admindata = config['dao'].get_admindata()
except BaseException as ex:
# Anzeige
print((f"L'erreur suivante s'est produite : {ex}"))
# Ende
sys.exit()
# Der vom Skript empfangene Parameter wird entfernt
sys.argv.pop()
# Die Testmethoden werden ausgeführt
print("tests en cours...")
unittest.main()
- Wir gehen nicht noch einmal auf die 11 Tests ein, die im Abschnitt |Test der Schicht [métier], Version 4| beschrieben sind;
- Zeilen 37–66: Wir werden das Testskript wie eine normale Anwendung ausführen und nicht als Test UnitTest. Erst in Zeile 66 wird das Framework UnitTest zum Einsatz kommen. In den vorherigen Tests haben wir die Methode [setUp] verwendet, um die Ausführung jedes einzelnen Tests zu konfigurieren. Wir haben dieselbe Konfiguration elfmal wiederholt, da die Funktion [setUp] vor jedem Test ausgeführt wird. Hier nehmen wir die Konfiguration einmal vor. Sie besteht darin, die globalen Variablen [métier] (Zeile 53) und [admindata] (Zeile 56) zu definieren, die anschließend von den Methoden von [TestDaoMétier] verwendet werden, beispielsweise in Zeile 12;
- Zeilen 39–47: Das Testskript erwartet einen Parameter [mysql / pgres], der angibt, ob die Datenbank MySQL oder PostgreSQL verwendet wird;
- Zeilen 50–51: Der Test wird konfiguriert;
- Zeile 53: Die Ebene [métier] wird aus der Konfiguration abgerufen;
- Zeile 56: Das Gleiche geschieht mit der Ebene [dao]. Dabei wird die Instanz [admindata] abgerufen, die die für die Steuerberechnung erforderlichen Daten enthält;
- Tests haben gezeigt, dass die Methode [unittest.main()] in Zeile 66 den vom Skript empfangenen Parameter [mysql / pgres] nicht ignorierte, ihm jedoch eine andere Bedeutung zuwies. Zeile 63 sorgt dafür, dass diese Methode keine Parameter mehr hat;
Wir erstellen zwei Ausführungskonfigurationen:
Wenn wir eine dieser beiden Konfigurationen ausführen, erhalten wir folgende Ergebnisse:
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
- Zeilen 5 und 7: Alle 11 Tests wurden erfolgreich bestanden;
Zur Erinnerung: Diese Tests überprüfen lediglich 11 Fälle der Steuerberechnung. Ihr erfolgreicher Abschluss kann dennoch ausreichen, um uns Vertrauen in die Schicht [dao] zu geben.
20.2.5. Das Hauptskript
Das Hauptskript [main] ist dasselbe wie in Version 4:
# Es wird auf einen MySQL- oder PostgreSQL-Parameter gewartet
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()
# Die Anwendung wird konfiguriert
import config
config = config.configure({'sgbd': sgbd})
# Der Syspath ist festgelegt – die Importe können durchgeführt werden
from ImpôtsError import ImpôtsError
# Die Anwendungsschichten werden abgerufen (sie sind bereits instanziiert)
dao = config["dao"]
métier = config["métier"]
try:
# Abruf der Steuerklassen
admindata = dao.get_admindata()
# Auslesen der Steuerzahlerdaten
taxpayers = dao.get_taxpayers_data()["taxpayers"]
# der Steuerzahler?
if not taxpayers:
raise ImpôtsError(57, f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
# Berechnung der Steuer der Steuerpflichtigen
for taxPayer in taxpayers:
# taxPayer ist sowohl ein Eingabe- als auch ein Ausgabeparameter
# taxPayer wird geändert
métier.calculate_tax(taxPayer, admindata)
# Schreiben der Ergebnisse in eine Textdatei
dao.write_taxpayers_results(taxpayers)
except ImpôtsError as erreur:
# Anzeige des Fehlers
print(f"L'erreur suivante s'est produite : {erreur}")
finally:
# Beendet
print("Travail terminé...")
Anmerkungen
- Zeilen 1–10: Der Parameter [mysql / pgres] wird abgerufen, der angibt, welches SGBD verwendet werden soll;
- Zeilen 12–14: Die Anwendung wird konfiguriert;
- Zeilen 16–17: Die Klasse [ImpôtsError] wird importiert. Diese wird in Zeile 38 benötigt;
- Zeilen 19–21: Es werden Referenzen auf die Schichten der Anwendung abgerufen;
- Zeile 25: Von der Schicht [dao] werden die Daten der Steuerbehörde angefordert. Die Schicht [métier] benötigt diese für die Steuerberechnung;
- Zeile 27: Die Daten (ID, Familienstand, Kinder, Gehalt) der Steuerzahler werden in eine Liste übernommen;
- Zeilen 29–30: Ist diese Liste leer, wird eine Ausnahme ausgelöst;
- Zeilen 32–35: Berechnung der Steuer für die Elemente der Liste [taxpayers];
- Zeile 37: Die Ergebnisse werden in die Datei jSON[résultats.json] geschrieben;
- Zeilen 38–40: Behandlung eines eventuellen Fehlers;
Für die Ausführung des Skripts werden zwei |Ausführungskonfigurationen| angelegt:
Die in der Datei [résultats.json] erhaltenen Ergebnisse stammen aus Version 4.
20.3. Anwendung 3: Steuerberechnung im interaktiven Modus
Wir stellen nun die Anwendung vor, mit der die Steuer interaktiv berechnet werden kann. Es handelt sich um eine Portierung von Anwendung 2 aus Version 4.
- Das Skript [main] startet den Dialog mit dem Benutzer mithilfe der Methode [ui.run] der Ebene [ui];
- die Schicht [ui]:
- nutzt die Schicht [dao], um die Daten für die Steuerberechnung abzurufen;
- fragt den Benutzer nach den Angaben zum Steuerpflichtigen, für den die Steuer berechnet werden soll;
- verwendet die Schicht [métier], um diese Berechnung durchzuführen;
Die Datei [config_layers] instanziiert eine zusätzliche Schicht:
def configure(config: dict) -> dict:
# Instanziierung der DAO-Schicht
from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
config["dao"] = ImpotsDaoWithAdminDataInDatabase(config)
# Instanziierung der Schicht [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)
# Konfiguration wird zurückgegeben
return config
Die Klasse [ImpôtsConsole], Zeilen 11–12, ist dieselbe wie in |Version 4|.
Das Hauptskript [main] lautet wie folgt:
# Es wird ein MySQL- oder PostgreSQL-Parameter erwartet
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()
# Die Anwendung wird konfiguriert
import config
config = config.configure({'sgbd': sgbd})
# Der Syspath ist konfiguriert – die Importe können durchgeführt werden
from ImpôtsError import ImpôtsError
# die Schicht wird abgerufen [ui]
ui = config["ui"]
# Code
try:
# Ausführung der Schicht [ui]
ui.run()
except ImpôtsError as ex1:
# Die Fehlermeldung wird angezeigt
print(f"L'erreur 1 suivante s'est produite : {ex1}")
except BaseException as ex2:
# Die Fehlermeldung wird angezeigt
print(f"L'erreur 2 suivante s'est produite : {ex2}")
finally:
# wird in jedem Fall ausgeführt
print("Travail terminé...")
- Zeilen 1–10: Das Skript erwartet einen Parameter [mysql / pgres], der angibt, welches SGBD verwendet werden soll;
- Zeilen 12–14: Die Anwendung wird konfiguriert;
- Zeilen 19–20: Die Ebene [ui] wird aus der Konfiguration abgerufen;
- Zeile 25: Sie wird ausgeführt;
Die Ergebnisse sind identisch mit denen von |Version 4|. Das konnte auch gar nicht anders sein, da alle Schnittstellen der Version 4 in der Version 5 beibehalten wurden.