Skip to content

20. Anwendungsübung: Version 5

Image

Wir werden drei Anwendungen entwickeln:

  • Anwendung 1 initialisiert die Datenbank, die die Datei [admindata.json] aus Version 4 ersetzen wird;
  • Anwendung 2 berechnet Steuern im Batch-Modus;
  • Anwendung 3 berechnet Steuern im interaktiven Modus;

20.1. Anwendung 1: Datenbankinitialisierung

Anwendung 1 wird die folgende Architektur aufweisen:

Image

Dies ist eine Weiterentwicklung der Architektur von Version 4 (siehe Abschnitt |Version 4|): Steuerdaten werden in einer Datenbank statt in einer JSON-Datei gespeichert. Die [DAO]-Schicht wird aktualisiert, um diese Änderung zu implementieren.

20.1.1. Die Datei [admindata.json]

Image

Die Datei [admindata.json] ist dieselbe wie in Version 4:


{
    "limites": [9964, 27519, 73779, 156244, 0],
    "coeffr": [0, 0.14, 0.3, 0.41, 0.45],
    "coeffn": [0, 1394.96, 5798, 13913.69, 20163.45],
    "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 aus diesem Wörterbuch als Spalten in der Datenbank verwenden.

20.1.2. Erstellen der Datenbanken

Wie im Abschnitt |Erstellen einer MySQL-Datenbank| gezeigt, erstellen wir eine MySQL-Datenbank mit dem Namen [dbimpots-2019], die dem Benutzer [admimpots] mit dem Passwort [mdpimpots] gehört. In [phpMyAdmin] sieht dies wie folgt aus:

Image

Ebenso erstellen wir, wie im Abschnitt |Erstellen einer PostgreSQL-Datenbank| gezeigt, eine PostgreSQL-Datenbank mit dem Namen [dbimpots-2019], deren Eigentümer der Benutzer [admimpots] ist und deren Passwort [mdpimpots] lautet. In [pgAdmin] sieht dies wie folgt aus:

Image

Die Datenbanken wurden erstellt, enthalten aber vorerst keine Tabellen. Diese werden vom [sqlalchemy]-ORM erstellt.

20.1.3. Von [sqlalchemy] zugeordnete Entitäten

Wir werden zwei Tabellen erstellen, um die Daten aus [admindata.json] zu kapseln:

Die von [sqlalchemy] definierte Tabelle [tbtranches] sammelt Daten aus den Arrays [limites, coeffr, coeffn] im Wörterbuch [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)
                           )

Die von [sqlalchemy] definierte Tabelle [tbconstantes] enthält die Konstanten aus dem Wörterbuch [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)
                             )

Die Entitäten, die diesen beiden Tabellen zugeordnet werden, sind wie folgt:

Image

Die Entität [Constants] kapselt die Konstanten aus dem Wörterbuch [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"]
  • Zeile 5: Die Klasse [Constants] erweitert die Klasse [BaseEntity];
  • Zeile 7: Über das [sqlalchemy]-Mapping erhält die Klasse [Constante] die Eigenschaft [_sa_instance_state]. Wir schließen sie aus dem [asdict]-Wörterbuch der Entität aus;
  • Zeilen 11–23: die Eigenschaften der Entität. Wir haben die im [admindata.json]-Wörterbuch verwendeten Namen wiederverwendet, um das Schreiben des Codes zu vereinfachen;

Die Entität [Tranche] kapselt eine Zeile aus den drei Arrays [limites, coeffr, coeffn] im Wörterbuch [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"]
  • Zeile 5: Die Klasse [Tranche] erweitert die Klasse [BaseEntity];
  • Zeile 7: Die von [sqlalchemy] hinzugefügte Eigenschaft [_sa_instance_state] wird aus dem [asdict]-Wörterbuch der Entität ausgeschlossen;
  • Zeilen 10–12: die Eigenschaften der Klasse;

Die Zuordnung zwischen den Entitäten [Constants, Slice] und den Tabellen [constants, slices] sieht wie folgt aus:

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)
  • Die Zuordnungen sind in den Zeilen 24–29 definiert. Wir haben die Zuordnungen zwischen den Eigenschaften der zugeordneten Entitäten und den Datenbanktabellen weggelassen. Dies ist möglich, wenn die Namen der Tabellenspalten mit denen der Eigenschaften übereinstimmen, denen sie zugeordnet werden sollen. Aus diesem Grund haben wir die Namen der Eigenschaften der zugeordneten Entitäten in die Tabellen aufgenommen. Dies erleichtert das Schreiben und Verstehen des Codes;

20.1.4. Die [sqlalchemy]-Konfigurationsdatei

Image

Wir haben soeben einen Teil der [sqlalchemy]-Konfiguration beschrieben. Die vollständige [config_database]-Datei lautet wie folgt:

 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
  • Zeile 1: Die Funktion [configure] erhält ein Wörterbuch als Parameter, dessen Schlüssel [dbms] angibt, welches DBMS verwendet werden soll: MySQL (mysql) oder PostgreSQL (pgres);
  • Zeilen 6–12: Die durch die Konfiguration angegebene Datenbank wird ausgewählt;
  • Zeilen 14–44: Entitäts-/Tabellenzuordnungen. Diese Zuordnungen sind einfach, da zwischen den Tabellen [tranches] und [constantes] keine Beziehung besteht. Sie sind unabhängig voneinander. Daher müssen keine Fremdschlüssel zwischen ihnen verwaltet werden;
  • Zeilen 46–51: Erstellen der Arbeitssitzung [session] der Anwendung;
  • Zeilen 53–58: Die relevanten Informationen werden in das Konfigurationswörterbuch geschrieben, das anschließend zurückgegeben wird;

20.1.5. Die [dao]-Schicht

Kehren wir zur Architektur der zu erstellenden Anwendung 1 zurück:

Image

Die [dao]-Schicht [1] muss die Datei [admindata.json] [2] lesen und deren Inhalt an eine der Datenbanken [3, 4] übertragen;

Image

Die [dao]-Schicht stellt die Schnittstelle [1] bereit und wird durch die Klasse [2] implementiert.

Die Schnittstelle [InterfaceDao4TransferAdminData2Database] sieht wie folgt aus:

 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
  • Zeilen 8–10: Die Schnittstelle definiert nur eine Methode [transfer_admindata_in_database] ohne Parameter. Da diese Methode Parameter benötigt (welche Datei?, welche Datenbank?), bedeutet dies, dass diese Parameter an den Konstruktor der Klassen übergeben werden, die diese Schnittstelle implementieren;

Die Klasse [DaoTransferAdminDataFromJsonFile2Database] implementiert die Schnittstelle [InterfaceDao4TransferAdminData2Database] wie folgt:

 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()
  • Zeile 13: Die Klasse [DaoTransferAdminDataFromJsonFile2Database] implementiert die Schnittstelle [InterfaceDao4TransferAdminData2Database];
  • Zeilen 15–17: Der Klassenkonstruktor nimmt das Konfigurationswörterbuch als Parameter entgegen. Die folgenden Schlüssel werden verwendet:
    • [admindataFilename] (Zeile 27): der Name der JSON-Datei, die die in die Datenbank zu übertragenden Steuerverwaltungsdaten enthält;
    • [database] Zeile 32: die [sqlalchemy]-Konfiguration der Anwendung;
  • Zeilen 34–37: Löschen der Tabellen [constants] und [brackets], falls vorhanden;
  • Zeilen 39–40: Neuerstellung der beiden Tabellen;
  • Zeile 43: Abrufen der [sqlalchemy]-Sitzung aus der Konfiguration;
  • Zeilen 45–51: Die Arrays [limits, coeffr, coeffn] aus dem Wörterbuch [admindata] werden der Sitzung hinzugefügt. Dazu werden Instanzen der Entität [Tranche] zur Sitzung hinzugefügt;
  • Zeilen 52–64: Eine Instanz der Entität [Constantes] wird der Sitzung hinzugefügt;
  • Zeilen 66–67: Die Sitzung wird validiert. Wenn die Sitzungsdaten noch nicht in der Datenbank vorhanden waren, werden sie an dieser Stelle eingefügt;
  • Zeilen 68–70: Fehlerbehandlung;
  • Zeilen 71–74: Die Sitzung wird geschlossen. Dies ist möglich, da die [dao]-Schicht nur einmal verwendet wird;

20.1.6. Anwendungskonfiguration

Image

Die Anwendung wird durch drei Dateien konfiguriert [1]:

  • [config] ist die allgemeine Konfigurationsdatei. Sie konfiguriert die [main]-Anwendung. Dabei wird sie von den beiden anderen Dateien unterstützt:
    • [config_database], die wir bereits betrachtet haben und die das ORM [sqlalchemy] konfiguriert;
    • [config_layers], die die Anwendungsschichten konfiguriert;

Die [config]-Datei sieht wie folgt aus:

 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
  • Zeilen 8–36: Erstellen Sie den Python-Pfad der Anwendung;
  • Zeilen 38–43: Fügen Sie den Pfad zur Datei [admindata.json] zur Konfiguration hinzu;
  • Zeilen 45–48: [SQLAlchemy]-Konfiguration;
  • Zeilen 50–53: Instanziierung der Anwendungsschichten;
  • Zeile 56: Rückgabe der allgemeinen Konfiguration;

Die Datei [config_layers] sieht wie folgt aus:

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
  • Zeilen 3–4: Instanziierung der [dao]-Schicht. Wir haben gesehen, dass der Konstruktor der Klasse [DaoTransferAdminDataFromJsonFile2Database] das allgemeine Konfigurationswörterbuch der Anwendung als Parameter erwartet;
  • Zeile 4: Der Verweis auf die [dao]-Schicht wird der Konfiguration hinzugefügt;
  • Zeile 7: Rückgabe der Konfiguration;

20.1.7. Das [main]-Skript der Anwendung

Image

Image

Das Hauptskript [main] lautet wie folgt:

 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é...")
  • Zeilen 1–10: Wir warten auf einen Parameter. Wir prüfen, ob er vorhanden und korrekt ist;
  • Zeilen 12–14: Wir konfigurieren die Anwendung (allgemein, SQLAlchemy, Layers), indem wir den gewählten DBMS-Typ als Parameter übergeben;
  • Zeilen 19–20: Wir benötigen die [dao]-Schicht. Wir rufen sie ab;
  • Zeile 25: Wir führen die Übertragung in die Datenbank durch. Alle von der Methode [transfer_admindata_in_database] benötigten Informationen sind in den Eigenschaften der [dao]-Schicht aus Zeile 20 verfügbar. Dort werden sie abgerufen;

Nach Ausführung des Skripts mit der MySQL-Datenbank enthält es die folgenden Elemente (phpMyAdmin):

Image

Image

Image

Spalte [3] zeigt die Werte, die MySQL dem Primärschlüssel [id] zugewiesen hat. Die Nummerierung beginnt bei 1. Der obige Screenshot wurde nach mehrmaliger Ausführung des Skripts aufgenommen.

Image

Image

Bei der PostgreSQL-Datenbank lauten die Ergebnisse wie folgt:

Image

  • Klicken Sie mit der rechten Maustaste auf [1] und anschließend auf [2-3];
  • In [4] werden die Daten zu den Steuerklassen übersichtlich angezeigt;

Das Gleiche machen wir für die Konstantentabelle [tbconstantes]:

Image

Image

Image

20.2. Anwendung 2: Steuerberechnung im Batch-Modus

Image

20.2.1. Architektur

Die Steuerberechnungsanwendung in Version 4 verwendete die folgende Architektur:

Image

Die [dao]-Schicht implementiert eine Schnittstelle [InterfaceImpôtsDao]. Wir haben eine Klasse erstellt, die diese Schnittstelle implementiert:

  • [TaxDaoWithAdminDataInJsonFile], die Steuerdaten aus einer JSON-Datei abrief. Das war Version 3;

Wir werden die Schnittstelle [InterfaceImpôtsDao] mithilfe einer neuen Klasse [ImpotsDaoWithTaxAdminDataInDatabase] implementieren, die Steuerverwaltungsdaten aus einer Datenbank abruft. Die [dao]-Schicht wird wie zuvor die Ergebnisse in eine JSON-Datei schreiben und Steuerzahlerdaten aus einer Textdatei abrufen. Wir wissen, dass die [business]-Schicht nicht geändert werden muss, wenn wir weiterhin die Schnittstelle [InterfaceImpôtsDao] einhalten.

Die neue Architektur sieht wie folgt aus:

Image

20.2.2. Anwendungskonfiguration

Image

Die Konfigurationsdatei [config_database] bleibt unverändert gegenüber Anwendung 1. Die Konfiguration [config] enthält neue Elemente:


    # é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"
    })
  • Zeilen 6–8: die absoluten Pfade der von Anwendung 2 verwendeten Textdateien;

Die Konfiguration der Ebenen [config_layers] entwickelt sich wie folgt:

 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
  • Zeilen 3–4: Die [dao]-Schicht wird nun durch die Klasse [TaxDaoWithAdminDataInDatabase] implementiert. Diese Klasse ist neu, implementiert jedoch dieselbe [DaoInterface]-Schnittstelle wie in Version 4 der Anwendungsübung;
  • Zeilen 7–8: Die [business]-Schicht wird von der Klasse [ImpôtsMétier] implementiert. Dies ist die Klasse, die in Version 4 der Anwendungsübung verwendet wird;

20.2.3. Die [DAO]-Schicht

Die Implementierungsklasse [ImpotsDaoWithAdminDataInDatabase] für die Schnittstelle [InterfaceImpôtsDao] sieht wie folgt aus:

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

Anmerkungen

  • Zeile 11: Die Klasse [ImpotsDaoWithAdminDataInDatabase] erbt von der in Version 4 vorgestellten Klasse [AbstractImpôtsDao]. Wir wissen, dass letztere die in derselben Version vorgestellte Schnittstelle [InterfaceDao] implementiert. Die Einhaltung dieser Schnittstelle ermöglicht es uns, die [business]-Schicht unverändert zu lassen;
  • Zeile 13: Der Klassenkonstruktor erhält das Konfigurationswörterbuch der Anwendung als Parameter;
  • Zeile 20: Die übergeordnete Klasse [] wird initialisiert. Sie implementiert teilweise die Schnittstelle [InterfaceDao]:
    • [get_taxpayers_data] liest die Datei [taxpayersdata.txt] mit den Steuerzahlerdaten;
    • [write_taxpayers_results] schreibt die Ergebnisse in die JSON-Datei [results.json];
    • [get_admindata] ist nicht implementiert;
  • Zeile 22: Die als Parameter übergebenen Konfigurationen werden gespeichert;
  • Zeile 27: Implementierung der Methode [get_admindata] der Schnittstelle [InterfaceDao]:
  • Zeilen 28–30: Die Methode [get_admindata] ruft Daten aus der Steuerverwaltung in ein Objekt vom Typ [AdminData] ab und speichert dieses Objekt in [self.__admindata]. Wird die Methode [get_admindata] mehrfach aufgerufen, wird die Datenbank nicht mehrfach abgefragt. Sie wird nur beim ersten Mal abgefragt. Bei nachfolgenden Aufrufen wird das Objekt [self.__admindata] zurückgegeben;
  • Zeilen 36–37: Abrufen der [sqlalchemy]-Sitzung, die während der Anwendungskonfiguration durch [config_database] erstellt wurde;
  • Zeile 40: Wir rufen die Steuerklassen in einer Liste ab;
  • Zeile 43: Wir rufen die Konstanten für die Steuerberechnung ab;
  • Zeile 46: Wir erstellen eine Instanz der Klasse [AdminData]. Zur Erinnerung: Diese leitet sich von [BaseEntity] ab;
  • Zeilen 48–54: Wir initialisieren die Arrays [limites, coeffr, coeffn] der [AdminData]-Instanz;
  • Zeilen 55–56: Initialisieren Sie die anderen Eigenschaften von [AdminData] mit den Konstanten für die Steuerberechnung. 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 [dao]-Schicht gespeichert, um bei nachfolgenden Aufrufen der Methode [get_admindata] zurückgegeben zu werden;
  • Zeile 60: Der vom aufrufenden Code angeforderte Wert wird zurückgegeben;
  • Zeilen 61–63: Fehlerbehandlung;
  • Zeilen 64–67: Die Datenbank wird nur einmal abgefragt. Wir können daher die [sqlalchemy]-Sitzung schließen;

20.2.4. Testen der [dao]-Schicht

In Version 4 dieser Anwendung haben wir eine Testklasse für die [business]-Schicht erstellt. Genauer gesagt testete sie sowohl die [business]- als auch die [DAO]-Schicht. Wir verwenden diesen Test erneut, um zu überprüfen, ob die [DAO]-Schicht wie erwartet funktioniert. Die [business]-Schicht bleibt jedoch unverändert.

Image

Image

Der [TestDaoMétier]-Test sieht wie folgt aus:


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()
  • Wir werden nicht erneut auf die 11 Tests eingehen, die im Abschnitt |[Business]-Layer-Test Version 4| beschrieben sind;
  • Zeilen 37–66: Wir führen das Testskript als normale Anwendung und nicht als UnitTest aus. Zeile 66 löst das UnitTest-Framework aus. In den vorherigen Tests haben wir die Methode [setUp] verwendet, um die Ausführung jedes Tests zu konfigurieren. Da die Funktion [setUp] vor jedem Test ausgeführt wird, haben wir dieselbe Konfiguration elfmal wiederholt. Hier führen wir die Konfiguration nur einmal durch. Sie besteht aus der Definition der globalen Variablen [business] in Zeile 53 und [admindata] in Zeile 56, die dann von den Methoden von [TestDaoBusiness] verwendet werden, zum Beispiel in Zeile 12;
  • Zeilen 39–47: Das Testskript erwartet einen Parameter [mysql / pgres], der angibt, ob eine MySQL- oder eine PostgreSQL-Datenbank verwendet wird;
  • Zeilen 50–51: Der Test wird konfiguriert;
  • Zeile 53: Die [business]-Schicht wird aus der Konfiguration abgerufen;
  • Zeile 56: Wir verfahren ebenso mit der [dao]-Schicht. Anschließend rufen wir die [admindata]-Instanz ab, die die zur Berechnung der Steuer erforderlichen Daten enthält;
  • Tests haben gezeigt, dass die Methode [unittest.main()] in Zeile 66 den an das Skript übergebenen Parameter [mysql / pgres] nicht ignorierte, sondern ihm stattdessen eine andere Bedeutung zuwies. Zeile 63 stellt sicher, dass diese Methode keine Parameter mehr hat;

Wir erstellen zwei Ausführungskonfigurationen:

Image

Image

Wenn wir eine dieser beiden Konfigurationen ausführen, erhalten wir die folgenden 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 bestanden;

Beachten Sie, dass diese Tests nur 11 Fälle der Steuerberechnung überprüfen. Ihr Erfolg kann dennoch ausreichen, um uns Vertrauen in die [dao]-Schicht zu geben.

20.2.5. Das Hauptskript

Image

Image

Das Hauptskript [main] ist dasselbe wie in Version 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é...")

Anmerkungen

  • Zeilen 1–10: Wir rufen den Parameter [mysql / pgres] ab, der das zu verwendende DBMS angibt;
  • Zeilen 12–14: Die Anwendung wird konfiguriert;
  • Zeilen 16–17: Die Klasse [ImpôtsError] wird importiert. Wir benötigen sie in Zeile 38;
  • Zeilen 19–21: Wir rufen Referenzen auf die Anwendungsschichten ab;
  • Zeile 25: Wir fordern die Daten der Steuerverwaltung aus der [dao]-Schicht an. Die [business]-Schicht benötigt diese zur Berechnung der Steuer;
  • Zeile 27: Wir rufen die Daten der Steuerzahler (ID, verheiratet, Kinder, Gehalt) in eine Liste ab;
  • Zeilen 29–30: Ist diese Liste leer, wird eine Ausnahme ausgelöst;
  • Zeilen 32–35: Berechnung der Steuer für die Elemente in der Liste [taxpayers];
  • Zeile 37: Die Ergebnisse werden in die JSON-Datei [results.json] geschrieben;
  • Zeilen 38–40: Behandlung etwaiger Fehler;

Um das Skript auszuführen, erstellen wir zwei |Ausführungskonfigurationen|:

Image

Die in der Datei [results.json] erhaltenen Ergebnisse stammen aus Version 4.

Image

20.3. Anwendung 3: Steuerberechnung im interaktiven Modus

Wir stellen nun die Anwendung vor, die eine interaktive Steuerberechnung ermöglicht. Es handelt sich um eine Portierung von Anwendung 2 aus Version 4.

Image

Image

  • Das [main]-Skript initiiert den Benutzerdialog mithilfe der Methode [ui.run] der [ui]-Schicht;
  • Die [ui]-Schicht:
    • nutzt die [dao]-Schicht, um die für die Steuerberechnung benötigten Daten abzurufen;
    • fragt den Benutzer nach Informationen zu dem Steuerpflichtigen, für den die Steuer berechnet werden soll;
    • nutzt die [business]-Schicht, um diese Berechnung durchzuführen;

Die Datei [config_layers] instanziiert eine zusätzliche Ebene:

 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

Die Klasse [ImpôtsConsole] in den Zeilen 11–12 ist dieselbe wie in |Version 4|.

Das Hauptskript [main] lautet wie folgt:

 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é...")
  • Zeilen 1–10: Das Skript erwartet einen Parameter [mysql / pgres], der das zu verwendende DBMS angibt;
  • Zeilen 12–14: Die Anwendung wird konfiguriert;
  • Zeilen 19–20: Die [ui]-Schicht wird aus der Konfiguration abgerufen;
  • Zeile 25: Sie wird ausgeführt;

Die Ergebnisse sind identisch mit denen von |Version 4|. Das kann auch gar nicht anders sein, da alle Schnittstellen aus Version 4 in Version 5 beibehalten wurden.

https://stahe.github.io
Dieses Tutorial, verfasst von Serge Tahé, wird der Öffentlichkeit unter den Bedingungen der Creative Commons Namensnennung – Nicht kommerziell – Weitergabe unter gleichen Bedingungen 3.0 Unported.