Skip to content

23. Übungsaufgabe: Version 6

23.1. Einführung

Wir kehren nun zu unserer Steuerberechnungsanwendung zurück. Wir werden verschiedene Webanwendungen darauf aufbauen.

In Version 5 unserer Anwendungsübung wurden die Daten der Steuerbehörde in einer Datenbank gespeichert. Diese Version 5 bestand aus zwei separaten Anwendungen, die gemeinsame Schichten nutzten:

  • eine Anwendung, die Steuern im |Batch|-Modus für Steuerzahler berechnete, die in einer Textdatei gespeichert waren;
  • eine Anwendung, die Steuern im |interaktiven| Modus für Steuerzahler berechnete, deren Daten über die Tastatur eingegeben wurden;

Die Version 5 der Anwendung zur Batch-Steuerberechnung wies folgende Architektur auf:

Image

Letztendlich wird die Webversion dieser Anwendung die folgende Architektur aufweisen:

Image

  • Der Web-Client [1] kommuniziert mit dem Web-Server [2], der wiederum mit dem DBMS [3] kommuniziert;
  • Der Webserver [2] behält die [Business]- [8]- und [DAO]- [9]-Schichten der ursprünglichen Anwendung bei;
  • Die ursprüngliche Anwendung behält ihr Hauptskript [4] und ihre [Business-]Schicht [15] bei. Die [Business-]Schichten [8] und [15] sind identisch;
  • die Client-Server-Kommunikation erfordert zwei zusätzliche Schichten:
    • die [Web]-Schicht [7], die die Webanwendung implementiert;
    • die [DAO]-Schicht [5], die als Client für die Webanwendung [7] fungiert;

In der endgültigen Version kann die Steuerberechnung im Batch-Verfahren auf zwei Arten durchgeführt werden:

  • Die Geschäftslogik für die Steuerberechnung wird von der [Business]-Schicht des Servers übernommen. Das [main]-Skript wird diese Methode verwenden;
  • Die Geschäftslogik für die Steuerberechnung wird von der [Business]-Schicht des Clients übernommen. Das Skript [main2] wird diese Methode verwenden;

Von nun an werden wir mehrere Client/Server-Anwendungen des oben beschriebenen Typs entwickeln, von denen jede eine oder mehrere neue Webentwicklungstechnologien veranschaulicht.

23.2. Der Webserver für die Steuerberechnung

23.2.1. Version 1

Image

Das Skript [server_01] ist die folgende Webanwendung:

Image

  • In [1] verwenden wir eine parametrisierte URL, in der wir drei Werte übergeben:
    • [married] (ja/nein), um anzugeben, ob der Steuerzahler verheiratet ist;
    • [children]: die Anzahl der Kinder des Steuerpflichtigen;
    • [salary]: das Jahresgehalt des Steuerzahlers;
  • In [2] gibt der Webserver eine JSON-Zeichenkette zurück, die den fälligen Steuerbetrag zusammen mit dessen verschiedenen Bestandteilen angibt;

Die Anwendungsarchitektur sieht wie folgt aus:

Image

  • Der Browser [1] fragt den Server [2] ab. Das Skript [server_01] implementiert die [Web]-Schicht [2] des Servers;
  • Die Schichten [3–8] sind diejenigen, die bereits in |Version 5| der Steuerberechnungsanwendung verwendet wurden. Wir verwenden sie unverändert wieder;
    • Die [Business]-Schicht [3] ist |hier| definiert;
    • die [DAO]-Schicht [4] ist |hier| definiert;

Die Webanwendung [server_01] wird mithilfe von drei Skripten konfiguriert:

  • [config], das die gesamte Anwendung konfiguriert;
  • [config_database], das den Datenbankzugriff konfiguriert. Wir werden mit den DBMS MySQL und PostgreSQL arbeiten;
  • [config_layers], das die Anwendungsschichten konfiguriert;

Das Skript [config] 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
def configure(config: dict) -> dict:
    import os

    #  step 1 ------
    #  folder of this file
    script_dir = os.path.dirname(os.path.abspath(__file__))
    #  root path
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
    #  absolute dependencies
    absolute_dependencies = [
        #  project files
        #  BaseEntity, MyException
        f"{root_dir}/classes/02/entities",
        #  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",
        #  ImpotsDaoWithAdminDataInDatabase
        f"{root_dir}/impots/v05/services",
        #  AdminData, ImpôtsError, TaxPayer
        f"{root_dir}/impots/v04/entities",
        #  Constants, slices
        f"{root_dir}/impots/v05/entities",
        #  IndexController
        f"{script_dir}/../controllers",
        #  scripts [config_database, config_layers]
        script_dir,
    ]
    #  set the syspath
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    #  step 2 ------
    #  application configuration
    #  list of users authorized to use the application
    config['users'] = [
        {
            "login": "admin",
            "password": "admin"
        }
    ]

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

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

    #  we return the configuration
    return config
  • Die Funktion [configure] nimmt ein [config]-Wörterbuch als Argument entgegen (Zeile 1) und gibt es als Ergebnis zurück (Zeile 54), nachdem sie dessen Inhalt erweitert hat. Man hätte schon längst darauf hinweisen können, dass es nicht notwendig war, das Ergebnis [config] zurückzugeben . Tatsächlich ist [config] eine Wörterbuchreferenz, die sich der aufrufende Code mit dem aufgerufenen Code teilt. Der aufrufende Code verfügt daher bereits über diese Referenz (Zeile 1), und es besteht keine Notwendigkeit, sie erneut zurückzugeben (Zeile 54). Daher wäre die folgende Schreibweise: 

config=[module].configure(config) (1)

ist überflüssig. Es reicht aus, zu schreiben:


[module].configure(config) (2)

Dennoch habe ich den Schreibstil (1) beibehalten, da ich dachte, dass er besser veranschaulicht, dass der aufgerufene Code das [config]-Wörterbuch modifiziert.

  • Zeile 1: Das von der Funktion [configure] empfangene [config]-Wörterbuch enthält einen Schlüssel „sgbd“, dessen Wert aus der Liste [„mysql“, „pgres“] entnommen wird. [mysql] bedeutet, dass die verwendete Datenbank von MySQL verwaltet wird, während „pgres“ bedeutet, dass die verwendete Datenbank von PostgreSQL verwaltet wird;
  • Zeilen 4–27: Wir listen alle Verzeichnisse auf, die für die Webanwendung notwendige Elemente enthalten. Sie werden Teil des Python-Pfads der Anwendung sein (Zeilen 30–31);
  • Zeilen 33–40: Nur bestimmte Benutzer dürfen auf die Anwendung zugreifen. Hier haben wir eine Liste mit einem einzigen Benutzer;
  • Zeilen 43–46: Das Skript [config_database] erstellt die Konfiguration für die verwendete Datenbank;
  • Zeile 46: Die vom Skript [config_database] erstellte Konfiguration ist ein Wörterbuch, das wir in der allgemeinen Konfiguration unter dem Schlüssel „database“ speichern;
  • Zeilen 48–51: Das Skript [config_layers] instanziiert die Schichten der Webanwendung. Es gibt ein Wörterbuch zurück, das in der allgemeinen Konfiguration unter dem Schlüssel „layers“ gespeichert wird;

Das Skript [config_database] ist dasjenige, das bereits in |Version 5| verwendet wurde. Wir fügen es hier zur Information ein:

 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
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 and rendered in a dictionary
    return {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
            "constantes_table": constantes_table, "session": session}

Das Skript [config_layers] konfiguriert die Webserver-Ebenen. Wir verwenden ein Skript wieder, das wir bereits kennen:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
def configure(config: dict) -> dict:
    #  instantiation of application layers

    #  dao
    from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
    dao = ImpotsDaoWithAdminDataInDatabase(config)

    #  business
    from ImpôtsMétier import ImpôtsMétier
    métier = ImpôtsMétier()

    #  put layer instances in a dictionary and return them to the calling code
    return {
        "dao": dao,
        "métier": métier
    }
  • Zeile 6: Die [dao]-Schicht wird mithilfe einer Datenbank implementiert;
  • [ImpotsDaoWithAdminDataInDatabase] wurde |hier| definiert;
  • [BusinessTaxes] wurde |hier| definiert;

Das Hauptskript [server_01] 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
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
#  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})

#  dependencies
from ImpôtsError import ImpôtsError
from TaxPayer import TaxPayer
import re
from flask import request
from myutils import json_response
from flask import Flask
from flask_api import status

#  data recovery from tax authorities
try:
    #  admindata will be read-only application data
    admindata = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
    print(f"L'erreur suivante s'est produite : {erreur}")
    sys.exit(1)

#  flask application
app = Flask(__name__)


#  Home URL : /?married=xx&children=yy&salary=zz
@app.route('/', methods=['GET'])
def index():
    #  initially no errors
    erreurs = []
    #  the query must have three parameters in the URL
    if len(request.args) != 3:
        erreurs.append("Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]")

    #  retrieve marital status in URL
    marié = request.args.get('marié')
    if marié is None:
        erreurs.append("paramètre [marié] manquant")
    else:
        marié = marié.strip().lower()
        erreur = marié != "oui" and marié != "non"
        if erreur:
            erreurs.append(f"paramétre marié [{marié}] invalide")

    #  retrieve the number of children in the URL
    enfants = request.args.get('enfants')
    if enfants is None:
        erreurs.append("paramètre [enfants] manquant")
    else:
        enfants = enfants.strip()
        match = re.match(r"^\d+", enfants)
        if not match:
            erreurs.append(f"paramétre enfants [{enfants}] invalide")
        else:
            enfants = int(enfants)

    #  the salary is retrieved from the URL
    salaire = request.args.get('salaire')
    if salaire is None:
        erreurs.append("paramètre [salaire] manquant")
    else:
        salaire = salaire.strip()
        match = re.match(r"^\d+", salaire)
        if not match:
            erreurs.append(f"paramétre salaire [{salaire}] invalide")
        else:
            salaire = int(salaire)

    #  invalid parameters in the URL?
    for key in request.args.keys():
        if key not in ['marié', 'enfants', 'salaire']:
            erreurs.append(f"paramètre [{key}] invalide")

    #  mistakes?
    if erreurs:
        #  an error response is sent to the client
        résultats = {"réponse": {"erreurs": erreurs}}
        return json_response(résultats, status.HTTP_400_BAD_REQUEST)

    #  no mistakes, we can work
    #  tAX CALCULATION
    taxpayer = TaxPayer().fromdict({'marié': marié, 'enfants': enfants, 'salaire': salaire})
    config["layers"]["métier"].calculate_tax(taxpayer, admindata)
    #  we send the response to the customer
    return json_response({"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK)


#  hand only
if __name__ == '__main__':
    #  start the Flask server
    app.config.update(ENV="development", DEBUG=True)
    app.run()
  • Zeilen 1–10: Abrufen des Parameters, der angibt, welches DBMS verwendet werden soll;
  • Zeilen 12–14: Mit diesen Informationen können wir die Anwendung konfigurieren. Insbesondere wird der Python-Pfad aufgebaut;
  • Zeilen 16–23: Mit dem neuen Python-Pfad importieren wir die erforderlichen Module;
  • Zeilen 25–31: Abrufen von Daten von der Steuerbehörde zur Berechnung der Steuer;
  • Zeilen 33–34: Instanziierung der Flask-Anwendung;
  • Zeile 38: Die Flask-Anwendung bedient nur die URL [/]. Sie erwartet eine URL im folgenden Format: [/ ?married=xx&children=yy&salary=zz], wobei:
    • xx: ja / nein;
    • yy: Anzahl der Kinder;
    • zz: Jahresgehalt;
  • Zeilen 40–89: Wir prüfen die Gültigkeit der URL-Parameter;
  • Zeile 41: Wir sammeln Fehlermeldungen in der Liste [errors];
  • Zeile 43: Wie du dich vielleicht erinnerst, befinden sich die Parameter der URL in [request.args] (siehe |hier|):
    • das [request]-Objekt ist das in Zeile 20 importierte Flask-Objekt;
    • das Objekt [request.args] verhält sich wie ein Wörterbuch;
  • Zeilen 43–44: Wir überprüfen, ob genau drei Parameter vorhanden sind (nicht weniger, nicht mehr);
  • Zeilen 46–49: Wir prüfen, ob der Parameter [married] in der URL vorhanden ist;
  • Zeilen 50–54: Wenn er vorhanden ist, prüfen wir, ob sein Wert in Kleinbuchstaben, ohne führende und nachfolgende Leerzeichen, „yes“ oder „no“ lautet;
  • Zeilen 56–59: Wir prüfen, ob der Parameter [children] in der URL enthalten ist;
  • Zeilen 60–66: Falls vorhanden, prüfen wir, ob sein Wert eine positive ganze Zahl ist;
  • Zeile 66: Beachten Sie, dass URL-Parameter und ihre Werte Zeichenfolgen sind. Der Wert des Parameters [children] wird in einen „int“ umgewandelt;
  • Zeilen 68–78: Für den Parameter [salary] führen wir dieselben Prüfungen durch wie für den Parameter [children];
  • Zeilen 81–83: Wir prüfen, ob in der URL keine anderen Parameter als [‘married’, ‘children’, ‘salary’] vorhanden sind;
  • Zeilen 85–89: Wenn nach all diesen Prüfungen die Liste [errors] nicht leer ist, senden wir diese Fehlerliste als JSON-Zeichenkette zusammen mit dem Statuscode [400 Bad Request] an den Client;

Da wir später häufig eine JSON-Zeichenkette als Antwort an den Client senden müssen, wurden die dafür erforderlichen wenigen Zeilen in das Modul [myutils.py] ausgelagert, das wir bereits verwendet haben:

Image

Das Skript [myutils.py] sieht nun 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
#  imports
import json
import os
import sys

from flask import make_response


def set_syspath(absolute_dependencies: list):
    #  absolute_dependencies: a list of absolute folder names

    ….


#  response generation HTTP jSON
def json_response(réponse: dict, status_code: int) -> tuple:
    #  response body HTTP
    response = make_response(json.dumps(réponse, ensure_ascii=False))
    #  response body HTTP is jSON
    response.headers['Content-Type'] = 'application/json; charset=utf-8'
    #  we send the HTTP response
    return response, status_code
  • Zeile 16: Die Funktion [json_response] erwartet zwei Parameter:
    • [response]: das Wörterbuch, das die an den Web-Client zu sendende JSON-Zeichenkette enthält;
    • [status_code]: der HTTP-Statuscode der Antwort;
  • Zeile 18: Wir legen den JSON-Body der Antwort fest;
  • Zeile 20: Wir fügen den HTTP-Header hinzu, der dem Web-Client mitteilt, dass er JSON erhalten wird;
  • Zeile 22: Wir senden die HTTP-Antwort an den aufrufenden Code. Es ist Aufgabe des aufrufenden Codes, diese an den Web-Client weiterzuleiten;

Die Datei [__init__.py] ändert sich wie folgt:


from .myutils import set_syspath, json_response

Die neue Version von [myutils] wird mit dem Befehl [pip install .] in einem PyCharm-Terminal unter den systemweiten Modulen installiert:


(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\packages>pip install .
Processing c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\packages
Using legacy setup.py install for myutils, since package 'wheel' is not installed.
Installing collected packages: myutils
  Attempting uninstall: myutils
    Found existing installation: myutils 0.1
    Uninstalling myutils-0.1:
      Successfully uninstalled myutils-0.1
    Running setup.py install for myutils ... done
Successfully installed myutils-0.1
  • Zeile 1: Sie müssen sich im Ordner [packages] befinden, um diesen Befehl einzugeben;

Der Code für das Skript [server_01] lautet wie folgt:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
…    
    #  mistakes?
    if erreurs:
        #  an error response is sent to the client
        résultats = {"réponse": {"erreurs": erreurs}}
        return json_response(résultats, status.HTTP_400_BAD_REQUEST)

    #  no mistakes, we can work
    #  tAX CALCULATION
    taxpayer = TaxPayer().fromdict({'id': 0, 'marié': marié, 'enfants': enfants, 'salaire': salaire})
    config["layers"]["métier"].calculate_tax(taxpayer, admindata)
    #  we send the response to the client
    return json_response({"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK)
  • Zeile 10: Zu diesem Zeitpunkt sind die erwarteten Parameter in der URL vorhanden und korrekt;
  • Zeile 10: Wir erstellen das [TaxPayer]-Objekt, das den Steuerzahler modelliert;
  • Zeile 11: Wir beauftragen die [business]-Schicht, die Steuer zu berechnen. Beachten Sie, dass die von der [business]-Schicht berechneten Elemente in das als Parameter übergebene [taxpayer]-Objekt eingefügt werden;
  • Zeile 13: Die Antwort wird als JSON-String an den Web-Client gesendet. Dies ist der JSON-String eines Dictionaries. Dem Schlüssel [result] ordnen wir das Dictionary des [taxpayer]-Objekts zu. Wir konnten das [taxpayer]-Objekt selbst nicht zuweisen, da es in JSON nicht serialisierbar ist;

Wir erstellen zwei Ausführungskonfigurationen, eine für MySQL und eine für PostgreSQL:

Image

Hier sind einige Ausführungsbeispiele (Sie haben die Anwendung [server_01] und das DBMS gestartet und dann über einen Browser die URL http://localhost:5000/ aufgerufen):

Image

Image

Hier ist ein Beispiel für die Anfrage in der Postman-Konsole:

Image


GET /?mari%C3%A9=xx&enfants=yy&salaire=zz HTTP/1.1
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: e4c5df8c-4bd6-4250-b789-b7b164db4eff
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive
 
HTTP/1.0 400 BAD REQUEST
Content-Type: application/json; charset=utf-8
Content-Length: 134
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, 17 Jul 2020 06:15:44 GMT
 
{"réponse": {"erreurs": ["paramètre marié [xx] invalide", "paramètre enfants [yy] invalide", "paramètre salaire [zz] invalide"]}}
  • Zeile 1: Es wird eine falsche URL angefordert;
  • Zeile 10: Der Server antwortet mit dem Status 400 BAD REQUEST;

23.2.2. Version 2

Image

In Version 2 des Servers wird die URL-Verarbeitung im Modul [index_controller] isoliert [5]:

 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
#  import dependencies
import re

from flask_api import status
from werkzeug.local import LocalProxy


#  URL set: /?married=xx&children=yy&salary=zz
def execute(request: LocalProxy, config: dict) -> tuple:
    #  dependencies
    from TaxPayer import TaxPayer

    #  initially no errors
    erreurs = []
    #  the query must have three parameters
    if len(request.args) != 3:
        erreurs.append("Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]")

    #  we retrieve the marital status of the URL
    marié = request.args.get('marié')
    if marié is None:
        erreurs.append("paramètre [marié] manquant")
    else:
        marié = marié.strip().lower()
        erreur = marié != "oui" and marié != "non"
        if erreur:
            erreurs.append(f"paramétre marié [{marié}] invalide")

    #  we retrieve the number of children in URL
    enfants = request.args.get('enfants')
    if enfants is None:
        erreurs.append("paramètre [enfants] manquant")
    else:
        enfants = enfants.strip()
        match = re.match(r"^\d+", enfants)
        if not match:
            erreurs.append(f"paramétre enfants {enfants} invalide")
        else:
            enfants = int(enfants)

    #  we recover the URL salary
    salaire = request.args.get('salaire')
    if salaire is None:
        erreurs.append("paramètre [salaire] manquant")
    else:
        salaire = salaire.strip()
        match = re.match(r"^\d+", salaire)
        if not match:
            erreurs.append(f"paramétre salaire {salaire} invalide")
        else:
            salaire = int(salaire)

    #  other parameters in the URL?
    for key in request.args.keys():
        if not key in ['marié', 'enfants', 'salaire']:
            erreurs.append(f"paramètre [{key}] invalide")

    #  mistakes?
    if erreurs:
        #  an error response is sent to the client
        résultats = {"réponse": {"erreurs": erreurs}}
        return résultats, status.HTTP_400_BAD_REQUEST

    #  no mistakes, we can work
    #  tAX CALCULATION
    taxpayer = TaxPayer().fromdict({'marié': marié, 'enfants': enfants, 'salaire': salaire})
    config["layers"]["métier"].calculate_tax(taxpayer, config["admindata"])
    #  we send the response to the customer
    return {"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK
  • Zeile 9: Die Funktion [execute] erhält zwei Parameter:
    • [request]: die HTTP-Anfrage des Clients;
    • [config]: das Konfigurationswörterbuch der Anwendung;

Das Skript [server_02] 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
#  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})

#  dependencies
from ImpôtsError import ImpôtsError
from flask import request
from myutils import json_response
from flask import Flask
import index_controller

#  data recovery from tax authorities
try:
    #  admindata will be read-only application data
    config['admindata'] = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
    print(f"L'erreur suivante s'est produite : {erreur}")
    sys.exit(1)

#  flask application
app = Flask(__name__)


#  Home URL : /?married=xx&child=yy&salary=zz
@app.route('/', methods=['GET'])
def index():
    #  execute the query
    résultat, statusCode = index_controller.execute(request, config)
    #  we send the answer
    return json_response(résultat, statusCode)


#  hand only
if __name__ == '__main__':
    #  start the server
    app.config.update(ENV="development", DEBUG=True)
    app.run()
  • Zeilen 36–41: Behandlung der Route /;
  • Zeile 39: Verwendung der Funktion [IndexController.execute];

Wir werden nun diese Technik anwenden: Jede Route wird von einem eigenen Modul verarbeitet.

Die Ausführungsergebnisse sind dieselben wie bei Version 1.

23.2.3. Version 3

Version 3 führt das Konzept der Authentifizierung ein.

Das Skript [server_03] sieht nun 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
#  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})

#  dependencies
from ImpôtsError import ImpôtsError
from flask import request
from myutils import json_response
from flask import Flask
from flask_httpauth import HTTPBasicAuth
import index_controller

#  data recovery from tax authorities
try:
    #  config['admindata'] will be read-only application data
    config["admindata"] = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
    print(f"L'erreur suivante s'est produite : {erreur}")
    sys.exit(1)

#  authentication manager
auth = HTTPBasicAuth()


#  authentication method
@auth.verify_password
def verify_credentials(login: str, password: str) -> bool:
    #  user list
    users = config['users']
    #  browse this list
    for user in users:
        if user['login'] == login and user['password'] == password:
            return True
    #  we didn't find
    return False


#  flask application
app = Flask(__name__)


#  Home URL : /?married=xx&child=yy&salary=zz
@app.route('/', methods=['GET'])
@auth.login_required
def index():
    #  execute the query
    résultat, statusCode = index_controller.execute(request, config)
    #  we send the answer
    return json_response(résultat, statusCode)


#  hand only
if __name__ == '__main__':
    #  start the server
    app.config.update(ENV="development", DEBUG=True)
    app.run()
  • Zeile 21: Importieren eines Authentifizierungshandlers. Es gibt verschiedene Arten der Authentifizierung für einen Webserver. Die hier verwendete heißt [HTTP Basic]. Jede Art der Authentifizierung folgt einem bestimmten Client-Server-Dialog;
  • Zeile 33: Erstellen einer Instanz des Authentifizierungshandlers;
  • Zeile 37: Die Annotation [@auth.verify_password] kennzeichnet die Funktion, die ausgeführt werden soll, wenn der Authentifizierungshandler den vom Client gemäß dem [HTTP Basic]-Protokoll gesendeten Benutzernamen und das Passwort überprüfen möchte;
  • Zeile 55: Die Annotation [@auth.login_required] kennzeichnet eine Route, für die der Web-Client authentifiziert werden muss. Wenn der Web-Client seine Anmeldedaten noch nicht gesendet hat, fordert der Webserver diese automatisch über das HTTP-Basic-Protokoll an;

Das Modul [flask_httpauth] muss installiert sein:


(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-servers\01\flask>pip install flask_httpauth
Collecting flask_httpauth
  Downloading Flask_HTTPAuth-4.1.0-py2.py3-none-any.whl (5.8 kB)
Requirement already satisfied: Flask in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from flask_httpauth) (1.1.2)
Requirement already satisfied: itsdangerous>=0.24 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (1.1.0)
Requirement already satisfied: click>=5.1 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (7.1.2)
Requirement already satisfied: Jinja2>=2.10.1 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (2.11.2)
Requirement already satisfied: Werkzeug>=0.15 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (1.0.1)
Requirement already satisfied: MarkupSafe>=0.23 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Jinja2>=2.10.1->Flask->flask_httpauth) (1.1.1
)
Installing collected packages: flask-httpauth
Successfully installed flask-httpauth-4.1.0

Schauen wir mal, was in der Postman-Konsole passiert. Sie:

  • Erstellen Sie eine Laufkonfiguration;
  • Starten Sie die Webanwendung;
  • Starten Sie die Datenbank Ihrer Wahl;
  • Rufen Sie die URL [/] mit Postman auf;

Der Client-Server-Dialog in der Postman-Konsole sieht wie folgt aus:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
GET / HTTP/1.1
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: e65e2a28-4fe3-423b-88b3-b3e5a83092b1
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive

HTTP/1.0 401 UNAUTHORIZED
Content-Type: text/html; charset=utf-8
Content-Length: 19
WWW-Authenticate: Basic realm="Authentication Required"
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, 17 Jul 2020 07:05:37 GMT

Unauthorized Access
  • Zeile 10: Der Server antwortet, dass wir nicht berechtigt sind, auf die URL [/] zuzugreifen;
  • Zeile 13: Es teilt uns mit, welches Authentifizierungsprotokoll wir verwenden sollen, in diesem Fall das Basic-Authentifizierungsprotokoll;

Es ist möglich, Postman so zu konfigurieren, dass die Benutzeranmeldedaten gemäß dem Basic-Authentifizierungsprotokoll gesendet werden:

Image

  • In [6-7] geben wir die Anmeldedaten ein, die im Skript [config] vorhanden sind: Image 

    config['users'] = [
        {
            "login": "admin",
            "password": "admin"
        }
    ]

Der Client-Server-Dialog in der Postman-Konsole sieht nun wie folgt aus:


GET / HTTP/1.1
Authorization: Basic YWRtaW46YWRtaW4=
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: 5ce20822-e87c-4eef-a2f4-b9eaec38d881
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive
 
HTTP/1.0 400 BAD REQUEST
Content-Type: application/json; charset=utf-8
Content-Length: 203
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, 17 Jul 2020 07:20:01 GMT
 
{"réponse": {"erreurs": ["Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]", "paramètre [marié] manquant", "paramètre [enfants] manquant", "paramètre [salaire] manquant"]}}
  • Zeile 2: Der Postman-Client sendet die Benutzeranmeldedaten [admin / admin] in verschlüsselter Form;
  • Zeile 17: Der Server antwortet korrekt. Er meldet Fehler, da die Parameter [verheiratet, Kinder, Gehalt] nicht gesendet wurden (Zeile 1), meldet jedoch keinen Authentifizierungsfehler;

Rufen wir nun die URL / über einen Browser auf (unten Firefox):

Image

  • Wie bei Postman hat Firefox die HTTP-Antwort vom Server mit den folgenden HTTP-Headern erhalten: 
1
2
3
4
HTTP/1.0 401 UNAUTHORIZED
…
WWW-Authenticate: Basic realm="Authentication Required"
…

Firefox stoppt, wie andere Browser auch, den Dialog nicht, wenn er diese Header erhält. Er fordert den Benutzer auf, die vom Server angeforderten Anmeldedaten einzugeben. Im obigen Beispiel erhält man durch einfache Eingabe von „admin / admin“ die Antwort des Servers:

Image

23.3. Der Web-Client des Steuerberechnungsservers

23.3.1. Einleitung

Im vorherigen Abschnitt war der Web-Client für den Steuerberechnungsserver ein Browser. In diesem Abschnitt wird der Web-Client ein Konsolenskript sein. Die Architektur sieht dann wie folgt aus:

Image

  • Der Web-Client besteht aus den Schichten [1–2];
  • der Webserver besteht aus den Schichten [3–9]. Wie im vorherigen Abschnitt erwähnt;

müssen wir daher die Schichten [1-2] schreiben.

Die Schicht [dao] [2] muss mit dem Webserver [3] kommunizieren können. Wir verstehen nun das HTTP-Protokoll und könnten beispielsweise mit dem bereits behandelten Modul [pycurl] ein Skript schreiben, das mit dem Webserver [3] kommuniziert. Es gibt jedoch Module, die auf die HTTP-Client-Server-Kommunikation spezialisiert sind. Wir werden eines davon verwenden, das Modul [requests]:


(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-servers\01\flask>pip install requests
Collecting requests
  Downloading requests-2.24.0-py2.py3-none-any.whl (61 kB)
     || 61 kB 137 kB/s
Collecting idna<3,>=2.5
  Downloading idna-2.10-py2.py3-none-any.whl (58 kB)
     || 58 kB 692 kB/s
Collecting chardet<4,>=3.0.2
  Downloading chardet-3.0.4-py2.py3-none-any.whl (133 kB)
     || 133 kB 1.3 MB/s
Collecting urllib3!=1.25.0,!=1.25.1,<1.26,>=1.21.1
  Downloading urllib3-1.25.9-py2.py3-none-any.whl (126 kB)
     || 126 kB 1.1 MB/s
Collecting certifi>=2017.4.17
  Downloading certifi-2020.6.20-py2.py3-none-any.whl (156 kB)
     || 156 kB 1.1 MB/s
Installing collected packages: idna, chardet, urllib3, certifi, requests
Successfully installed certifi-2020.6.20 chardet-3.0.4 idna-2.10 requests-2.24.0 urllib3-1.25.9

Die Verzeichnisstruktur für die Web-Client-Skripte sieht wie folgt aus:

Image

Das Skript implementiert die in |Version 1| beschriebene Anwendung zur Steuerberechnung im Batch-Modus. Die neueste Version dieser Anwendung ist |Version 5|. Hier noch einmal zur Erinnerung, wie sie funktioniert:

  • Die Steuerzahler, für die die Steuer berechnet wird, sind in der Textdatei [taxpayersdata.txt] aufgeführt: 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
# données valides : id, marié, enfants, salaire
1,oui,2,55555
2,oui,2,50000
3,oui,3,50000
4,non,2,100000
5,non,3,100000
6,oui,3,100000
7,oui,5,100000
8,non,0,100000
9,oui,2,30000
10,non,0,200000
11,oui,3,200000
# on crée des lignes erronées
# pas assez de valeurs
11,12
# des valeurs erronées
x,x,x,x
  • Die Ergebnisse werden in zwei Dateien gespeichert:
  • Die Textdatei [errors.txt] listet die in der Steuerzahlerdatei festgestellten Fehler auf: 

Analyse du fichier C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-clients\01\main/../data/input/taxpayersdata.txt
 
Ligne 15, not enough values to unpack (expected 4, got 2)
Ligne 17, MyException[1, L'identifiant d'une entité <class 'TaxPayer.TaxPayer'> doit être un entier >=0]
  • (Fortsetzung)
    • Die JSON-Datei [results.json] enthält die Ergebnisse der Steuerberechnung für die verschiedenen Steuerzahler: 

[
  {
    "id": 0,
    "marié": "oui",
    "enfants": 2,
    "salaire": 55555,
    "impôt": 2814,
    "surcôte": 0,
    "taux": 0.14,
    "décôte": 0,
    "réduction": 0
  },
  {
    "id": 1,
    "marié": "oui",
    "enfants": 2,
    "salaire": 50000,
    "impôt": 1384,
    "surcôte": 0,
    "taux": 0.14,
    "décôte": 384,
    "réduction": 347
  },
  …
]

23.3.2. Konfiguration des Web-Clients

Image

Die Konfiguration erfolgt mithilfe von zwei Skripten:

  • [config], das alle Konfigurationen außerhalb der Architekturschichten übernimmt;
  • [config_layers], das die Konfiguration der Architekturschichten übernimmt;

Das Skript [config] 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
59
60
def configure(config: dict) -> dict:
    import os

    #  step 1 ------

    #  folder of this file
    script_dir = os.path.dirname(os.path.abspath(__file__))

    #  root path
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"

    #  absolute dependencies
    absolute_dependencies = [
        #  project files
        #  BaseEntity, MyException
        f"{root_dir}/classes/02/entities",
        #  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",
        #  ImpotsDaoWithAdminDataInDatabase
        f"{root_dir}/impots/v05/services",
        #  AdminData, ImpôtsError, TaxPayer
        f"{root_dir}/impots/v04/entities",
        #  Constants, slices
        f"{root_dir}/impots/v05/entities",
        #  ImpôtsDaoWithHttpClient
        f"{script_dir}/../services",
        #  configuration scripts
        script_dir,
    ]

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

    #  step 2 ------
    #  application configuration with constants
    config.update({
        "taxpayersFilename": f"{script_dir}/../data/input/taxpayersdata.txt",
        "resultsFilename": f"{script_dir}/../data/output/résultats.json",
        "errorsFilename": f"{script_dir}/../data/output/errors.txt",
        "server": {
            "urlServer": "http://127.0.0.1:5000/",
            "authBasic": True,
            "user": {
                "login": "admin",
                "password": "admin"
            }
        }
    }
    )

    #  step 3 ------
    #  layer instantiation
    import config_layers
    config['layers'] = config_layers.configure(config)

    #  we return the configuration
    return config
  • Zeile 1: Die Funktion [configure] nimmt als Parameter das Wörterbuch entgegen, das mit Konfigurationsinformationen gefüllt werden soll. Dieses Wörterbuch kann bereits vorbelegt oder leer sein. Hier ist es leer;
  • Zeilen 40–42: die absoluten Pfade der drei Textdateien, die von der [dao]-Schicht verwaltet werden;
  • Zeilen 43–50: dem Schlüssel [server] zugeordnet, die Informationen, die die [dao]-Schicht über den Webserver benötigt, mit dem sie kommunizieren muss:
    • Zeile 44: die URL des Webdienstes;
    • Zeile 45: Der Schlüssel [authBasic] wird auf „True“ gesetzt, wenn der Zugriff auf die URL eine Basic-Authentifizierung erfordert;
    • Zeilen 46–49: die Anmeldedaten des Benutzers, der sich authentifizieren wird, falls eine Authentifizierung erforderlich ist;
  • Zeilen 56–57: Wir instanziieren die Schichten – in diesem Fall die einzelne [dao]-Schicht – und platzieren die Schichtenreferenzen in [config] unter dem Schlüssel [layers];

Das Skript [config_layers] lautet wie folgt:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
def configure(config: dict) -> dict:
    #  instantiation of applicatuon layers

    #  dao layer
    from ImpôtsDaoWithHttpClient import ImpôtsDaoWithHttpClient
    dao = ImpôtsDaoWithHttpClient(config)

    #  make the layer configuration
    return {
        "dao": dao
    }
  • Zeile 1: Die Funktion [configure] erhält das Wörterbuch, das die Anwendung konfiguriert;
  • Zeilen 4–6: Die [dao]-Schicht wird instanziiert. In Zeile 6 übergeben wir ihr die Anwendungskonfiguration, in der sie die benötigten Informationen findet;
  • Zeilen 8–11: Es wird ein Wörterbuch zurückgegeben, das den Verweis auf die [dao]-Schicht enthält;

23.3.3. Das Hauptskript [main]

Das Hauptskript [main] ist eine Variante des Skripts aus |Version 5|:

 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
#  configure the application
import config
config = config.configure({})

#  dependencies
from ImpôtsError import ImpôtsError

#  code
try:
    #  we recover the [dao] layer
    dao = config["layers"]["dao"]
    #  reading taxpayer data
    taxpayers = dao.get_taxpayers_data()["taxpayers"]
    #  taxpayers?
    if not taxpayers:
        raise ImpôtsError(f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
    #  tax calculation
    for taxpayer in taxpayers:
        #  taxpayer is both an input and output parameter
        #  taxpayer will be modified
        dao.calculate_tax(taxpayer)
    #  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é...")
  • Zeilen 2–3: Die Anwendung wird konfiguriert;
  • Zeile 13: Die [dao]-Schicht liefert die Liste der Steuerzahler, für die Steuern berechnet werden müssen;
  • Zeile 21: Die [dao]-Schicht berechnet die Steuer für jeden einzelnen;
  • Zeile 23: Die Ergebnisse werden in einer JSON-Datei gespeichert;

23.3.4. Implementierung der [dao]-Schicht

Image

Werfen wir noch einmal einen Blick auf die verwendete Client/Server-Architektur:

Image

  • In [2, 6] sehen wir, dass die [dao]-Schicht zwei Aufgaben hat:
    • Sie greift auf das Dateisystem zu, um sowohl Steuerzahlerdaten zu lesen als auch die Ergebnisse der Steuerberechnungen zu schreiben. Wir verfügen bereits über eine |AbstractImpôtsDao|-Klasse, die dies leisten kann. Sie ist seit |Version 4| im Einsatz;
    • sie kommuniziert mit dem Webserver [3];

In |Version 5| kommunizierte das Hauptskript [main] [1] direkt mit der [Business]-Schicht [4]. Wir möchten dieses Skript lieber nicht ändern. Um dies zu erreichen, stellen wir sicher, dass die [DAO]-Schicht [2] die Schnittstelle der [business]-Schicht [4] implementiert. Auf diese Weise scheint das Hauptskript [main] direkt mit der [business]-Schicht [4] zu kommunizieren und kann die Tatsache, dass sich diese auf einem anderen Rechner befindet, vollständig ignorieren.

Eine Definition der Klasse, die die [DAO]-Schicht [2] implementiert, könnte wie folgt lauten:


class ImpôtsDaoWithHttpClient(AbstractImpôtsDao, InterfaceImpôtsMétier):
  • Die Klasse [TaxDaoWithHttpClient]:
    • erbt von der Klasse [AbstractTaxDao], wodurch sie die Kommunikation mit dem Dateisystem [6] handhaben kann;
    • implementiert die Schnittstelle [InterfaceImpôtsMétier], um das Hauptskript [main] von |Version 5| nicht ändern zu müssen;

Der vollständige Code für die Klasse [TaxDaoWithHttpClient] 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
#  imports
import requests
from flask_api import status

from AbstractImpôtsDao import AbstractImpôtsDao
from AdminData import AdminData
from ImpôtsError import ImpôtsError
from InterfaceImpôtsMétier import InterfaceImpôtsMétier
from TaxPayer import TaxPayer


class ImpôtsDaoWithHttpClient(AbstractImpôtsDao, InterfaceImpôtsMétier):

    #  manufacturer
    def __init__(self, config: dict):
        #  parent initialization
        AbstractImpôtsDao.__init__(self, config)
        #  parameter memory
        self.__config_server = config["server"]

    #  unused [AbstractImpôtsDao] method
    def get_admindata(self) -> AdminData:
        pass

    #  tAX CALCULATION
    def calculate_tax(self: object, taxpayer: TaxPayer, admindata: AdminData = None):
        #  we let the exceptions rise
        #  get parameters
        params = {"marié": taxpayer.marié, "enfants": taxpayer.enfants, "salaire": taxpayer.salaire}
        #  connection with Auth Basic authentication?
        if self.__config_server['authBasic']:
            response = requests.get(
                #  URL of the queried server
                self.__config_server['urlServer'],
                #  URL parameters
                params=params,
                #  basic authentication
                auth=(
                    self.__config_server["user"]["login"],
                    self.__config_server["user"]["password"]))
        else:
            #  connection without Auth Basic authentication
            response = requests.get(self.__config_server['urlServer'], params=params)
        #  check
        print(response.text)
        #  response status code HTTP
        status_code = response.status_code
        #  we put the response jSON in a dictionary
        résultat = response.json()
        #  error if status code other than 200 OK
        if status_code != status.HTTP_200_OK:
            #  we know that the errors have been associated with the [errors] key in the response
            raise ImpôtsError(87, résultat['réponse']['erreurs'])
        #  we know that the result has been associated with the [result] key in the response
        #  modify the input parameter with this result
        taxpayer.fromdict(résultat["réponse"]["result"])
  • Zeilen 21–23: Die Klasse [AbstractTaxDao] (Zeile 12) verfügt über eine abstrakte Methode [get_admindata]. Wir müssen diese implementieren, auch wenn wir sie nicht verwenden (admindata wird vom Server verwaltet, nicht vom Client);
  • Zeile 26: Die Methode [calculate_tax] gehört zur Schnittstelle [InterfaceImpôtsMétier] (Zeile 12). Wir müssen sie implementieren;
  • Zeile 15: Der Konstruktor erhält das Konfigurationswörterbuch der Anwendung als einzigen Parameter;
  • Zeilen 16–17: Die übergeordnete Klasse [AbstractTaxDao] wird initialisiert, indem ihr – auch hier – die Anwendungskonfiguration übergeben wird. Dort findet sie die Namen der drei Textdateien, die sie verwalten muss;
  • Zeilen 18–19: Informationen zum Webserver für die Steuerberechnung werden lokal innerhalb der Klasse gespeichert;
  • Zeile 26: Die Methode [calculate_tax] erhält ein Objekt vom Typ |Taxpayer| als Parameter. Um der Signatur der Methode [InterfaceImpôtsMétier.calculate_tax] zu entsprechen, erhält sie zudem einen Parameter [admindata], der die Daten der Steuerverwaltung kapseln soll. Auf der Client-Seite verfügen wir nicht über diese Daten. Dieser Parameter bleibt immer [None]. Diese Umgehungslösung deutet darauf hin, dass die Klasse [ImpôtsMétier] ursprünglich schlecht konzipiert war:
  • Die Signatur von [calculate_tax] hätte einfach lauten sollen: 

def calculate_tax(self, taxpayer: TaxPayer)

und der Parameter [admindata: AdminData] hätte an den Klassenkonstruktor übergeben werden sollen;

  • Zeile 27: Der Code für die Methode [calculate_tax] wurde nicht in einen try/catch/finally-Block gekapselt. Das bedeutet, dass etwaige Ausnahmen nicht abgefangen werden und an den aufrufenden Code, in diesem Fall das Skript [main], weitergeleitet werden. Dieses Skript fängt alle Ausnahmen ab, die aus der [dao]-Schicht weitergeleitet werden;
  • Zeile 28: Die Steuerberechnung erfolgt serverseitig. Wir müssen daher mit dem Server kommunizieren. Dazu verwenden wir das in Zeile 2 importierte [requests]-Modul;
  • Zeilen 31–43: Um eine GET-Anfrage an den Webserver zu senden, verwenden wir die Methode [requests.get]:
    • Zeilen 33–34: Der erste Parameter der Methode ist die URL, die aufgerufen werden soll;
    • Zeilen 35–40: Die beiden anderen Parameter sind benannte Parameter, deren Reihenfolge keine Rolle spielt;
    • Zeilen 35–36: Der Wert des benannten Parameters [params] muss ein Wörterbuch sein, das die Informationen enthält, die in der URL in der Form [/url?param1=value1&param2=value2&…] enthalten sein sollen;
    • Zeile 29: Das Wörterbuch enthält die drei Parameter [married, children, salary], die der Webserver erwartet. Wir müssen uns keine Gedanken über die Kodierung (sogenanntes URL-Encoding) machen, die diese Parameter durchlaufen müssen. [requests] übernimmt dies;
    • Zeilen 37–40: Der Parameter [auth] ist ein Tupel aus zwei Elementen (login, password). Er stellt die Anmeldedaten für die Basic-Authentifizierung dar;
  • Zeilen 44–45: Diese beiden Zeilen dienen nur zu Lehrzwecken (wir werden sie auskommentieren, sobald das Debugging abgeschlossen ist):
    • [response] steht für die HTTP-Antwort des Servers;
    • [response.text] stellt den Text des in dieser Antwort enthaltenen Dokuments dar. Während des Debuggens ist es nützlich, zu überprüfen, was der Server uns gesendet hat;
  • Zeile 47: [response.status_code] ist der HTTP-Statuscode der empfangenen Antwort. Unser Server sendet nur drei:
    • 200 OK
    • 400 BAD REQUEST
    • 500 INTERNALER SERVERFEHLER
  • Zeile 49: Unser Server sendet immer JSON, auch im Falle eines Fehlers. Die Funktion [response.json()] erstellt aus der empfangenen JSON-Zeichenkette ein Dictionary. Sehen wir uns die beiden möglichen Formen der JSON-Zeichenkette an: 

{"réponse": {"erreurs": ["Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]", "paramètre [marié] manquant", "paramètre [enfants] manquant", "paramètre [salaire] manquant"]}}

{"réponse": {"result": {"id": 0, "marié": "oui", "enfants": 3, "salaire": 200000, "impôt": 42842, "surcôte": 17283, "taux": 0.41, "décôte": 0, "réduction": 0}}}
  • Zeilen 51–53: Wenn der Statuscode nicht 200 ist, wird eine Ausnahme mit den in der Antwort enthaltenen Fehlermeldungen ausgelöst;
  • Zeile 56: Das durch die Steuerberechnung erzeugte Wörterbuch abrufen und damit den Eingabeparameter [taxpayer] aktualisieren;

23.3.5. Ausführung

So führen Sie den Client aus:

  • Starten Sie den Server [server_03] mit dem DBMS Ihrer Wahl;
  • Führen Sie das Skript [main] des Clients aus;

Die Ergebnisse finden Sie im Ordner [data/output]. Sie entsprechen denen der Version 5.

23.4. Tests der [dao]-Schicht

Kehren wir zur Client-Server-Anwendungsarchitektur zurück:

Image

  • Im Client-Code haben wir sichergestellt, dass die [dao]-Schicht [1] dieselbe Schnittstelle bereitstellt wie die [business]-Schicht [3]. Wir werden daher die Testklasse |TestDaoMétier| verwenden, die wir zuvor behandelt haben, um die [business]-Schicht [3] zu testen;

Die Testklasse wird in der folgenden Umgebung ausgeführt:

Image

  • Die Konfiguration [2] ist identisch mit der Konfiguration [1], die wir gerade betrachtet haben;

Die Testklasse [TestHttpClientDao] 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
import unittest


class TestHttpClientDao(unittest.TestCase):

    def test_1(self) -> None:
        from TaxPayer import TaxPayer

        #  { 'married': 'yes', 'children': 2, 'salary': 55555,
        #  tax': 2814, 'surcôte': 0, 'décôte': 0, 'réduction': 0, 'taux': 0.14}
        taxpayer = TaxPayer().fromdict({"marié": "oui", "enfants": 2, "salaire": 55555})
        dao.calculate_tax(taxpayer)
        #  check
        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

        #  { 'married': 'yes', 'children': 3, 'salary': 200000,
        #  tax': 42842, 'surcôte': 17283, 'décôte': 0, 'réduction': 0, 'taux': 0.41}
        taxpayer = TaxPayer().fromdict({'marié': 'oui', 'enfants': 3, 'salaire': 200000})
        dao.calculate_tax(taxpayer)
        #  checks
        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__':

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

    #  dao layer
    dao = config['layers']['dao']

    #  test methods are executed
    print("tests en cours...")
    unittest.main()

Diese Klasse ähnelt derjenigen, die bereits in Version 4 der Anwendung behandelt wurde.

  • Zeilen 40–41: Konfiguration der Testumgebung;
  • Zeile 44: Wir rufen eine Referenz auf die [DAO]-Schicht ab;
  • Zeilen 47–48: Wir führen die Tests aus;

Um die Tests auszuführen, erstellen wir eine |Ausführungskonfiguration|:

Image

  • Wir erstellen eine Ausführungskonfiguration für ein Konsolenskript, nicht für einen UnitTest;

Beim Ausführen dieser Konfiguration werden die folgenden Ergebnisse erzielt:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
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/http-clients/01/tests/TestHttpClientDao.py
tests en cours...
{"réponse": {"result": {"marié": "oui", "enfants": 2, "salaire": 55555, "impôt": 2814, "surcôte": 0, "taux": 0.14, "décôte": 0, "réduction": 0}}}
....{"réponse": {"result": {"marié": "non", "enfants": 0, "salaire": 200000, "impôt": 64210, "surcôte": 7498, "taux": 0.45, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 3, "salaire": 200000, "impôt": 42842, "surcôte": 17283, "taux": 0.41, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 2, "salaire": 50000, "impôt": 1384, "surcôte": 0, "taux": 0.14, "décôte": 384, "réduction": 347}}}
{"réponse": {"result": {"marié": "oui", "enfants": 3, "salaire": 50000, "impôt": 0, "surcôte": 0, "taux": 0.14, "décôte": 720, "réduction": 0}}}
...{"réponse": {"result": {"marié": "non", "enfants": 2, "salaire": 100000, "impôt": 19884, "surcôte": 4480, "taux": 0.41, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "non", "enfants": 3, "salaire": 100000, "impôt": 16782, "surcôte": 7176, "taux": 0.41, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 3, "salaire": 100000, "impôt": 9200, "surcôte": 2180, "taux": 0.3, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 5, "salaire": 100000, "impôt": 4230, "surcôte": 0, "taux": 0.14, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "non", "enfants": 0, "salaire": 100000, "impôt": 22986, "surcôte": 0, "taux": 0.41, "décôte": 0, "réduction": 0}}}
....
{"réponse": {"result": {"marié": "oui", "enfants": 2, "salaire": 30000, "impôt": 0, "surcôte": 0, "taux": 0.0, "décôte": 0, "réduction": 0}}}
----------------------------------------------------------------------
Ran 11 tests in 0.130s

OK

Process finished with exit code 0

Alle 11 Tests bestanden.

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.