Introduction à Python 3 et au framework web Flask par l'exemple

Le script [server_01] est l’application web suivante :

• en [1], on utilise une URL paramétrée dans laquelle on passe trois valeurs :

  • [marié] (oui / non) pour indiquer si le contribuable est marié ;

  • [enfants] : le nombre d’enfants du contribuable ;

  • [salaire] : salaire annuel du contribuable ;

• en [2], le serveur web renvoie une chaîne jSON qui donne le montant de l’impôt à payer avec avec ses différentes composantes ;

L’architecture de l’application est la suivante :

• le navigateur [1] interroge le serveur [2]. Le script [server_01] implémente la couche [web] [2] du serveur ;

• les couches [3-8] sont celles déjà utilisées dans la version 5 de l’application du calcul de l’impôt. Nous les reprenons telles-quelles ;

  • la couche [métier] [3] est définie ici ;

  • la couche [dao] [4] est définie ici ;

L’application web [server_01] est configurée à l’aide de trois scripts :

• [config] qui configure la totalité de l’application ;

• [config_database] qui configure l’accès à la base de données. On travaillera avec les SGBD MySQL et PostgreSQL ;

• [config_layers] qui configure les couches de l’application ;

Le script [config] est le suivant :

def configure(config: dict) -> dict:
    import os

    # étape 1 ------
    # dossier de ce fichier
    script_dir = os.path.dirname(os.path.abspath(__file__))
    # chemin racine
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
    # dépendances absolues
    absolute_dependencies = [
        # dossiers du projet
        # 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",
        # Constantes, tranches
        f"{root_dir}/impots/v05/entities",
        # IndexController
        f"{script_dir}/../controllers",
        # scripts [config_database, config_layers]
        script_dir,
    ]
    # on fixe le syspath
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    # étape 2 ------
    # configuration de l'application
    # liste des utilisateurs autorisés à utiliser l'application
    config['users'] = [
        {
            "login": "admin",
            "password": "admin"
        }
    ]

    # étape 3 ------
    # configuration base de données
    import config_database
    config["database"] = config_database.configure(config)

    # étape 4 ------
    # instanciation des couches de l'application
    import config_layers
    config['layers'] = config_layers.configure(config)

    # on rend la configuration
    return config

• la fonction [configure] reçoit un dictionnaire [config] en paramètre (ligne 1) et le rend comme résultat (ligne 54) après avoir enrichi son contenu. On aurait pu dire depuis longtemps qu’il n’était pas nécessaire de rendre le résultat [config]. En effet [config] est une référence de dictionnaire que le code appelant partage avec le code appelé. Le code appelant possède donc déjà cette référence (ligne 1) et il est inutile de la lui redonner (ligne 54). Ainsi écrire :

   

  Sélectionnez

  1.
  

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

  est redondant. Il suffit d’écrire :

   

  Sélectionnez

  1.
  

  [module].configure(config) (2)
  

  Néanmoins, j’ai gardé le type (1) d’écriture parce que j’ai pensé qu’il montrait peut-être mieux, que le code appelé modifiait le dictionnaire [config].

• ligne 1 : le dictionnaire [config] reçu par la fonction [configure] a une clé ‘sgbd’ qui prend sa valeur dans la liste ['mysql', 'pgres']. [mysql] signifie que la base de données utilisée est gérée par MySQL alors que ‘pgres’ signifie que la base de données utilisée est gérée par PostgreSQL ;

• lignes 4-27 : on liste tous les dossiers contenant des éléments nécessaires à l’application web. Ils feront partie du Python Path de l’application (lignes 30-31) ;

• lignes 33-40 : on ne va autoriser que certains utilisateurs à accéder à l’application. Ici on a une liste avec un unique utilisateur ;

• lignes 43-46 : c’est le script [config_database] qui construit la configuration de la base de données utilisée ;

• ligne 46 : la configuration construite par le script [config_database] est un dictionnaire qu’on range dans la configuration générale associé à la clé ‘database’ ;

• lignes 48-51 : le script [config_layers] instancie les couches de l’application web. Il rend un dictionnaire qu’on range dans la configuration générale associé à la clé ‘layers’ ;

Le script [config_database] est celui déjà utilisé dans la version 5. On le redonne pour mémoire :

def configure(config: dict) -> dict:
    # configuration sqlalchemy
    from sqlalchemy import create_engine, Table, Column, Integer, MetaData, Float
    from sqlalchemy.orm import mapper, sessionmaker

    # chaînes de connexion aux bases de données exploitées
    connection_strings = {
        'mysql': "mysql+mysqlconnector://admimpots:mdpimpots@localhost/dbimpots-2019",
        'pgres': "postgresql+psycopg2://admimpots:mdpimpots@localhost/dbimpots-2019"
    }
    # chaîne de connexion à la base de données exploitée
    engine = create_engine(connection_strings[config['sgbd']])

    # metadata
    metadata = MetaData()

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

    # 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)
                           )
    # les mappings
    from Tranche import Tranche
    mapper(Tranche, tranches_table)

    from Constantes import Constantes
    mapper(Constantes, constantes_table)

    # la session factory
    session_factory = sessionmaker()
    session_factory.configure(bind=engine)

    # une session
    session = session_factory()

    # on enregistre certaines informations et on les rend dans un dictionnaire
    return {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
            "constantes_table": constantes_table, "session": session}

Le script [config_layers] configure les couches du serveur web. On reprend un script déjà rencontré :

def configure(config: dict) -> dict:
    # instanciation des couches de l'application
    
    # dao
    from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
    dao = ImpotsDaoWithAdminDataInDatabase(config)
    
    # métier
    from ImpôtsMétier import ImpôtsMétier
    métier = ImpôtsMétier()

    # on met les instances de couches dans un dictionnaire qu'on rend au code appelant
    return {
        "dao": dao,
        "métier": métier
    }

• ligne 6 : la couche [dao] est implémentée avec une base de données ;

• [ImpotsDaoWithAdminDataInDatabase] a été définie ici ;

• [ImpôtsMétier] a été définie ici ;

Le script principal [server_01] est le suivant :

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

# dépendances
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

# récupération des données de l'administration fiscale
try:
    # admindata sera une donnée de portée application en lecture seule
    admindata = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
    print(f"L'erreur suivante s'est produite : {erreur}")
    sys.exit(1)

# application Flask
app = Flask(__name__)


# Home URL : /?marié=xx&enfants=yy&salaire=zz
@app.route('/', methods=['GET'])
def index():
    # au départ pas d'erreurs
    erreurs = []
    # la requête doit avoir trois paramètres dans l’URL
    if len(request.args) != 3:
        erreurs.append("Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]")

    # on récupère le statut marital dans l’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")

    # on récupère le nombre d'enfants dans l’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)

    # on récupère le salaire dans l’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)

    # des paramètres invalides dans l’URL ?
    for key in request.args.keys():
        if key not in ['marié', 'enfants', 'salaire']:
            erreurs.append(f"paramètre [{key}] invalide")

    # des erreurs ?
    if erreurs:
        # on envoie une réponse d'erreur au client
        résultats = {"réponse": {"erreurs": erreurs}}
        return json_response(résultats, status.HTTP_400_BAD_REQUEST)

    # pas d'erreurs, on peut travailler
    # calcul de l'impôt
    taxpayer = TaxPayer().fromdict({'marié': marié, 'enfants': enfants, 'salaire': salaire})
    config["layers"]["métier"].calculate_tax(taxpayer, admindata)
    # on envoie la réponse au client
    return json_response({"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK)


# main uniquement
if __name__ == '__main__':
    # on lance le serveur Flask
    app.config.update(ENV="development", DEBUG=True)
    app.run()

• lignes 1-10 : on récupère le paramètre qui indique quel SGBD utiliser ;

• lignes 12-14 : munie de cette information on peut configurer l’application. Le Python Path est notamment construit ;

• lignes 16-23 : muni du nouveau Python Path, on importe les éléments dont on a besoin ;

• lignes 25-31 : on récupère les données de l’administration fiscale qui permettent de faire le calcul de l’impôt ;

• lignes 33-34 : instanciation de l’application Flask ;

• ligne 38 : l’application Flask ne sert que l’URL [/]. Elle attend une URL paramétrée de la façon suivante [/ ?marié=xx&enfants=yy&salaire=zz] avec :

  • xx : oui / non ;

  • yy : nombre d’enfants ;

  • zz : salaire annuel ;

• lignes 40-89 : on vérifie la validité des paramètres de l’URL ;

• ligne 41 : on va cumuler les messages d’erreur dans la liste [erreurs] ;

• ligne 43 : on se rappelle peut-être que les paramètres de l’URL paramétrée sont retrouvés dans [request.args] (voir ici) :

  • l’objet [request] est l’objet Flask importé ligne 20 ;

  • l’objet [request.args] se comporte comme un dictionnaire ;

• lignes 43-44 : on vérifie qu’on a trois paramètres exactement (pas moins, pas plus) ;

• lignes 46-49 : on vérifie que le paramètre [marié] est présent dans l’URL ;

• lignes 50-54 : s’il est présent on vérifie que sa valeur minuscule débarrassée des ‘blancs’ de début / fin est oui ou non ;

• lignes 56-59 : on vérifie que le paramètre [enfants] est dans l’URL ;

• lignes 60-66 : s’il est présent on vérifie que sa valeur est un entier positif ;

• ligne 66 : il ne faut pas oublier que les paramètres de l’URL et leurs valeurs sont des chaînes de caractères. La valeur du paramètre [enfants] est transformée en ‘int’ ;

• lignes 68-78 : pour le paramètre [salaire], on fait les mêmes tests que pour le paramètre [enfants] ;

• lignes 81-83 : on vérifie qu’il n’y a pas de paramètres autres que ['marié', 'enfants', 'salaire'] dans l’URL ;

• lignes 85-89 : si après toutes ces vérifications, la liste [erreurs] n’est pas vide alors on envoie cette liste d’erreurs au client sous la forme d’une chaîne jSON  et le code de statut [400 Bad Request] ;

Comme par la suite, nous aurons souvent l’occasion d’envoyer une chaîne jSON en réponse au client, les quelques lignes nécessaires à cet envoi ont été factorisées dans le module [myutils.py] que nous avons déjà utilisé :

Le script [myutils.py] devient le suivant :

# imports
import json
import os
import sys

from flask import make_response


def set_syspath(absolute_dependencies: list):
    # absolute_dependencies : une liste de noms absolus de dossiers

    ….


# génération d'une réponse HTTP jSON
def json_response(réponse: dict, status_code: int) -> tuple:
    # corps de la réponse HTTP
    response = make_response(json.dumps(réponse, ensure_ascii=False))
    # corps de la réponse HTTP est du jSON
    response.headers['Content-Type'] = 'application/json; charset=utf-8'
    # on envoie la réponse HTTP
    return response, status_code

• ligne 16 : la fonction [json_response] attend deux paramètres :

  • [réponse] : le dictionnaire dont il faut envoyer la chaîne jSON au client web ;

  • [status_code] : le code de statut HTTP de la réponse ;

• ligne 18 : on fixe le corps jSON de la réponse ;

• ligne 20 : on ajoute l’entête HTTP qui dit au client web qu’il va recevoir du jSON ;

• ligne 22 : on envoie la réponse HTTP au code appelant. A charge pour lui de l’envoyer au client web ;

Le fichier [__init__.py] évolue de la façon suivante :

from .myutils import set_syspath, json_response

La nouvelle version de [myutils] est installée parmi les modules de portée machine avec la commande [pip install .] dans un terminal Pycharm :

(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

• ligne 1 : il faut être dans le dossier [packages] pour taper cette instruction ;

Le code du script [server_01] se poursuit de la façon suivante :

…    
    # des erreurs ?
    if erreurs:
        # on envoie une réponse d'erreur au client
        résultats = {"réponse": {"erreurs": erreurs}}
        return json_response(résultats, status.HTTP_400_BAD_REQUEST)

    # pas d'erreurs, on peut travailler
    # calcul de l'impôt
    taxpayer = TaxPayer().fromdict({'id': 0, 'marié': marié, 'enfants': enfants, 'salaire': salaire})
    config["layers"]["métier"].calculate_tax(taxpayer, admindata)
    # on envoie la réponse au client
    return json_response({"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK)

• ligne 10 : lorsqu’on est là, les paramètres attendus dans l’URL sont là et sont corrects ;

• ligne 10 : on crée l’objet [TaxPayer] qui modélise le contribuable ;

• ligne 11 : on demande à la couche [métier] de calculer l’impôt. On rappelle que les éléments calculés par la couche [métier] sont insérés dans l’objet [taxpayer] passé en paramètre ;

• ligne 13 : la réponse est envoyée au client web sous la forme d’une chaîne jSON. Celle-ci est la chaîne jSON d’un dictionnaire. Associé à la clé [result], on y met le dictionnaire de l’objet [taxpayer]. On ne pouvait pas mettre l’objet [taxpayer] lui-même car celui-ci n’est pas sérialisable en jSON ;

On crée deux configurations d’exécution, une pour MySQL, l’autre pour PostgreSQL :

Voici quelques exemples d’exécution (vous avez lancé l’application [server_01] et le SGBD utilisé puis vous demandez l’URL http://localhost:5000/ avec un navigateur) :

Voici un exemple d’exécution dans la console de Postman :

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"]}}

• ligne 1 : une URL incorrecte est demandée ;

• ligne 10 : le serveur répond avec le statut 400 BAD REQUEST ;

La version 2 du serveur isole le traitement de l’URL dans le module [index_controller] [5] :

# import des dépendances
import re

from flask_api import status
from werkzeug.local import LocalProxy


# URL paramétrée : /?marié=xx&enfants=yy&salaire=zz
def execute(request: LocalProxy, config: dict) -> tuple:
    # dépendances
    from TaxPayer import TaxPayer

    # au départ pas d'erreurs
    erreurs = []
    # la requête doit avoir trois paramètres
    if len(request.args) != 3:
        erreurs.append("Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]")

    # on récupère le statut marital de l'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")

    # on récupère le nombre d'enfants de l'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)

    # on récupère le salaire de l'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)

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

    # des erreurs ?
    if erreurs:
        # on envoie une réponse d'erreur au client
        résultats = {"réponse": {"erreurs": erreurs}}
        return résultats, status.HTTP_400_BAD_REQUEST

    # pas d'erreurs, on peut travailler
    # calcul de l'impôt
    taxpayer = TaxPayer().fromdict({'marié': marié, 'enfants': enfants, 'salaire': salaire})
    config["layers"]["métier"].calculate_tax(taxpayer, config["admindata"])
    # on envoie la réponse au client
    return {"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK

• ligne 9 : la fonction [execute] reçoit deux paramètres :

  • [request] : la requête HTTP du client ;

  • [config] : le dictionnaire de configuration de l’application ;

Le script [server_02] est le suivant :

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

# dépendances
from ImpôtsError import ImpôtsError
from flask import request
from myutils import json_response
from flask import Flask
import index_controller

# récupération des données de l'administration fiscale
try:
    # admindata sera une donnée de portée application en lecture seule
    config['admindata'] = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
    print(f"L'erreur suivante s'est produite : {erreur}")
    sys.exit(1)

# application Flask
app = Flask(__name__)


# Home URL : /?marié=xx&enfant=yy&salaire=zz
@app.route('/', methods=['GET'])
def index():
    # on exécute la requête
    résultat, statusCode = index_controller.execute(request, config)
    # on envoie la réponse
    return json_response(résultat, statusCode)


# main uniquement
if __name__ == '__main__':
    # on lance le serveur
    app.config.update(ENV="development", DEBUG=True)
    app.run()

• ligne 36-41 : le traitement de la route / ;

• ligne 39 : utilisation de la fonction [IndexController.execute] ;

Nous utiliserons désormais cette technique : chaque route sera traitée par un module qui lui sera propre.

Les résultats d’exécution sont les mêmes que pour la version 1.

La version 3 introduit la notion d’authentification.

Le script [server_03] devient le suivant :

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

# dépendances
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

# récupération des données de l'administration fiscale
try:
    # config['admindata'] sera une donnée de portée application en lecture seule
    config["admindata"] = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
    print(f"L'erreur suivante s'est produite : {erreur}")
    sys.exit(1)

# gestionnaire d'authentification
auth = HTTPBasicAuth()


# méthode d'authentification
@auth.verify_password
def verify_credentials(login: str, password: str) -> bool:
    # liste des utilisateurs
    users = config['users']
    # on parcourt cette liste
    for user in users:
        if user['login'] == login and user['password'] == password:
            return True
    # on n'a pas trouvé
    return False


# application Flask
app = Flask(__name__)


# Home URL : /?marié=xx&enfant=yy&salaire=zz
@app.route('/', methods=['GET'])
@auth.login_required
def index():
    # on exécute la requête
    résultat, statusCode = index_controller.execute(request, config)
    # on envoie la réponse
    return json_response(résultat, statusCode)


# main uniquement
if __name__ == '__main__':
    # on lance le serveur
    app.config.update(ENV="development", DEBUG=True)
    app.run()

• ligne 21 : on importe un gestionnaire d’authentification. Il existe divers types d’authentification auprès d’un serveur web. Celle que nous utilisons ici s’appelle [HTTP Basic]. Chaque type d’authentification suit un dialogue client / serveur précis ;

• ligne 33 : on crée une instance du gestionnaire d’authentification ;

• ligne 37 : l’annotation [@auth.verify_password] tague la fonction à exécuter lorsque le gestionnaire d’authentification veut vérifier les login / password transmis par le client selon le protocole [HTTP Basic] ;

• ligne 55 : l’annotation [@auth.login_required] tague une route pour laquelle le client web doit être authentifié. Si le client web n’a pas encore envoyé ses identifiants, le serveur web va automatiquement les lui demander selon le protocole HTTP basic ;

Le module [flask_httpauth] doit être installé :

(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

Voyons ce qui se passe avec la console Postman. Vous :

• créez une configuration d’exécution ;

• lancez l’application web ;

• lancez le SGBD de votre choix ;

• demandez l’URL [/] avec Postman ;

Le dialogue client / serveur dans la console Postman est le suivant :

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

• ligne 10 : le serveur répond que nous ne sommes pas autorisés à accéder à l’URL [/] ;

• ligne 13 : il nous indique le protocole d’authentification à utiliser, ici le protocole dit Authentification Basic ;

Il est possible de configurer Postman pour qu’il envoie les identifiants de l’utilisateur selon le protocole Auth Basic :

• en [6-7] nous mettons les identifiants présents dans le script [config] :

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

Le dialogue client / serveur dans la console Postman devient le suivant :

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"]}}

• ligne 2 : le client Postman envoie sous forme codée les identifiants de l’utilisateur [admin / admin] ;

• ligne 17 : le serveur répond correctement. Il signale des erreurs car on n’a pas envoyé les paramètres [marié, enfants, salaire] (ligne 1) mais il ne signale pas d’erreur d’authentification ;

Maintenant demandons l’URL / avec un navigateur (Firefox ci-dessous) :

• comme avec Postman, Firefox a reçu la réponse HTTP du serveur avec les entêtes HTTP :

HTTP/1.0 401 UNAUTHORIZED
…
WWW-Authenticate: Basic realm="Authentication Required"
…

Firefox comme d’autres navigateurs n’arrêtent pas le dialogue lorsqu’il reçoit ces entêtes. Il demandent à l’utilisateur les identifiants réclamés par le serveur. Il suffit ci-dessus de taper admin / admin pour recevoir la réponse du serveur :

Dans le paragraphe précédent, le client web du serveur de calcul de l’impôt était un navigateur. Dans cette partie, le client web sera un script console. L’architecture devient la suivante :

• le client web est formé des couches [1-2] ;

• le serveur web est formé des couches [3-9]. Il a été écrit dans le paragraphe précédent ;

Il nous faut donc écrire les couches [1-2].

La couche [dao] [2] doit savoir communiquer avec le serveur web [3]. Nous connaissons maintenant le protocole HTTP et nous pourrions écrire, avec le module [pycurl] déjà étudié par exemple, un script communiquant avec le serveur web [3]. Cependant il existe des modules spécialisés dans les dialogues HTTP client / serveur. Nous allons utiliser l’un d’entre eux, le module [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

L’arborescence des scripts du client web est la suivante :

Le script va implémenter l’application de calcul de l’impôt en mode batch décrite dès la version 1. La dernière version de cette application est la version 5. On rappelle son fonctionnement :

• les contribuables dont on va calculer l’impôt sont rassemblés dans le fichier texte [taxpayersdata.txt] :

# 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

• les résultats vont dans deux fichiers :

  • le fichier texte [errors.txt] rassemble les erreurs détectées dans le fichier des contribuables :

     

    Sélectionnez

    1.
    2.
    3.
    4.
    

    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]
    

  • le fichier jSON [résultats.json] rassemble les résultats des calculs de l’impôt des différents contribuables :

     

    Sélectionnez

    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.
    

    [
      {
        "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
      },
      …
    ]
    

La configuration se fait à l’aide de deux scripts :

• [config] qui assure la totalité de la configuration hors couches de l’architecture ;

• [config_layers] qui assure la configuration des couches de l’architecture ;

Le script [config] est le suivant :

def configure(config: dict) -> dict:
    import os

    # étape 1 ------

    # dossier de ce fichier
    script_dir = os.path.dirname(os.path.abspath(__file__))

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

    # dépendances absolues
    absolute_dependencies = [
        # dossiers du projet
        # 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",
        # Constantes, tranches
        f"{root_dir}/impots/v05/entities",
        # ImpôtsDaoWithHttpClient
        f"{script_dir}/../services",
        # scripts de configuration
        script_dir,
    ]

    # on fixe le syspath
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    # étape 2 ------
    # configuration de l'application avec des constantes
    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"
            }
        }
    }
    )

    # étape 3 ------
    # instanciation des couches
    import config_layers
    config['layers'] = config_layers.configure(config)

    # on rend la configuation
    return config

• ligne 1 : la fonction [configure] reçoit en paramètre le dictionnaire à remplir avec les informations de configuration. Celui-ci peut être déjà pré-rempli ou vide. Ici, il sera vide ;

• lignes 40-42 : les noms absolus des trois fichiers texte gérés par la couche [dao] ;

• lignes 43-50 : associées à la clé [server], les informations que doit connaître la couche [dao] sur le serveur web avec lequel elle doit communiquer :

  • ligne 44 : l’URL du service web ;

  • ligne 45 : la clé [authBasic] vaut True si l’accès à l’URL nécessite une authentification de type Basic ;

  • lignes 46-49 : les identifiants de l’utilisateur qui va s’authentifier si l’authentification est demandée ;

• lignes 56-57 : on instancie les couches, ici l’unique couche [dao] et on met les références des couches dans [config] associées à la clé [layers] ;

Le script [config_layers] est le suivant :

def configure(config: dict) -> dict:
    # instanciation des couches de l'applicatuon

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

    # on rend la configuation des couches
    return {
        "dao": dao
    }

• ligne 1 : la fonction [configure] reçoit le dictionnaire qui configure l’application ;

• lignes 4-6 : la couche [dao] est instanciée. Ligne 6, on lui passe la configuration de l’application dans lequel elle trouvera les informations dont elle a besoin ;

• lignes 8-11 : on rend un dictionnaire dans lequel on a mis la référence de la couche [dao] ;

Le script principal [main] est une variante de celui de la version 5 :

# on configure l'application
import config
config = config.configure({})

# dépendances
from ImpôtsError import ImpôtsError

# code
try:
    # on récupère la couche [dao]
    dao = config["layers"]["dao"]
    # lecture des données des contribuables
    taxpayers = dao.get_taxpayers_data()["taxpayers"]
    # des contribuables ?
    if not taxpayers:
        raise ImpôtsError(f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
    # calcul de l'impôt des contribuables
    for taxpayer in taxpayers:
        # taxpayer est à la fois un paramètre d'entrée et de sortie
        # taxpayer va être modifié
        dao.calculate_tax(taxpayer)
    # écriture des résultats dans un fichier texte
    dao.write_taxpayers_results(taxpayers)
except ImpôtsError  as erreur:
    # affichage de l'erreur
    print(f"L'erreur suivante s'est produite : {erreur}")
finally:
    # terminé
    print("Travail terminé...")

• lignes 2-3 : l’application est configurée ;

• ligne 13 : la couche [dao] fournit la liste des contribuables dont on doit calculer l’impôt ;

• ligne 21 : la couche [dao] calcule l’impôt de chacun d’eux ;

• ligne 23 : les résultats sont enregistrés dans un fichier jSON ;

Revenons sur l’architecture client / serveur utilisée :

• en [2, 6], on voit que la couche [dao] a deux rôles :

  • elle accède au système de fichiers à la fois pour lire les données des contribuables et écrire les résultats des calculs de l’impôt. Nous avons déjà une classe AbstractImpôtsDao qui sait faire cela. Elle a été utilisée dès la version 4 ;

  • elle dialogue avec le serveur web [3] ;

Dans la version 5, le script principal [main] [1] dialoguait directement avec la couche [métier] [4]. On voudrait ne pas changer ce script. Pour cela, on va faire en sorte que la couche [dao] [2] implémente l’interface de la couche [métier] [4]. Ainsi le script principal [main] aura l’impression de dialoguer directement avec la couche [métier] [4] et pourra ignorer complètement que celle-ci se situe sur une autre machine.

Une définition de la classe implémentant la couche [dao] [2] pourrait être celle-ci :

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

• la classe [ImpôtsDaoWithHttpClient] :

  • hérite de la classe [AbstractImpôtsDao] ce qui va lui permettre de gérer le dialogue avec le système de fichiers [6] ;

  • implémente l’interface [InterfaceImpôtsMétier] pour ne pas avoir à changer le script principal [main] de la version 5 ;

Le code complet de la classe [ImpôtsDaoWithHttpClient] est le suivant :

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

    # constructeur
    def __init__(self, config: dict):
        # initialisation parent
        AbstractImpôtsDao.__init__(self, config)
        # mémorisation paramètres
        self.__config_server = config["server"]

    # méthode inutilisée de [AbstractImpôtsDao]
    def get_admindata(self) -> AdminData:
        pass

    # calcul de l'impôt
    def calculate_tax(self: object, taxpayer: TaxPayer, admindata: AdminData = None):
        # on laisse remonter les exceptions
        # paramètres du get
        params = {"marié": taxpayer.marié, "enfants": taxpayer.enfants, "salaire": taxpayer.salaire}
        # connexion avec authentification Auth Basic ?
        if self.__config_server['authBasic']:
            response = requests.get(
                # URL du serveur interrogé
                self.__config_server['urlServer'],
                # paramètres de l'URL
                params=params,
                # authentification Basic
                auth=(
                    self.__config_server["user"]["login"],
                    self.__config_server["user"]["password"]))
        else:
            # connexion sans authentification Auth Basic
            response = requests.get(self.__config_server['urlServer'], params=params)
        # vérification
        print(response.text)
        # code de statut de la réponse HTTP
        status_code = response.status_code
        # on met la réponse jSON dans un dictionnaire
        résultat = response.json()
        # erreur si code de statut différent de 200 OK
        if status_code != status.HTTP_200_OK:
            # on sait que les erreurs ont été associées à la clé [erreurs] de la réponse
            raise ImpôtsError(87, résultat['réponse']['erreurs'])
        # on sait que le résultat a été associé à la clé [result] de la réponse
        # on modifie le paramètre d'entrée avec ce résultat
        taxpayer.fromdict(résultat["réponse"]["result"])

• lignes 21-23 : la classe [AbstractImpôtsDao] (ligne 12) possède une méthode abstraite [get_admindata]. Nous sommes obligés de l’implémenter même si on ne s’en sert pas (admindata est géré par le serveur pas par le client) ;

• ligne 26 : la méthode [calculate_tax] appartient à l’interface [InterfaceImpôtsMétier] (ligne 12). il nous faut l’implémenter ;

• ligne 15 : le constructeur reçoit en unique paramètre le dictionnaire de la configuration de l’application ;

• lignes 16-17 : la classe parent [AbstractImpôtsDao] est initialisée en lui passant, là également, la configuration de l’application. Elle y trouvera les noms des trois fichiers texte qu’elle a à gérer ;

• lignes 18-19 : on mémorise localement dans la classe les informations concernant le serveur web de calcul de l’impôt ;

• ligne 26 : la méthode [calculate_tax] reçoit en paramètre un objet de type Taxpayer. Pour respecter la signature de la méthode [InterfaceImpôtsMétier.calculate_tax], elle reçoit également un paramètre [admindata] qui est censé encapsuler les données de l’administration fiscale. Côté client, on n’a pas ces données. Ce paramètre restera toujours à [None]. Cette contorsion fait dire que la classe [ImpôtsMétier] a été initialement mal écrite :

  • la signature de [calculate_tax] aurait dû être simplement :

     

    Sélectionnez

    1.
    

    def calculate_tax(self, taxpayer: TaxPayer)
    

  et le paramètre [admindata : AdminData] aurait dû être passé au constructeur de la classe ;

• ligne 27 : le code de la méthode [calculate_tax] n’a pas été encapsulé dans un try / catch / finally. Cela veut dire que les éventuelles exceptions ne seront pas gérées et remonteront au code appelant, en l’occurrence le script [main]. Celui-ci arrête bien toutes les exceptions remontant de la couche [dao] ;

• ligne 28 : le calcul de l’impôt se fait côté serveur. Il va donc falloir communiquer avec lui. On le fait avec le module [requests] importé ligne 2 ;

• lignes 31-43 : pour envoyer une requête GET au serveur web, on utilise la méthode [requests.get] :

  • lignes 33-34 : le 1^er paramètre de la méthode est l’URL à contacter ;

  • lignes 35-40 : les deux autres paramètres sont des paramètres nommés dont l’ordre n’importe pas ;

  • lignes 35-36 : la valeur du paramètre nommé [params] doit être un dictionnaire contenant les informations à mettre dans l’URL sous la forme [/url ?param1=valeur1&param2=valeur2&…] ;

    • ligne 29 : le dictionnaire contenant les trois paramètres [marié, enfants, salaire] que le serveur web attend. On n’a pas à s’occuper de l’encodage (appelé urlencoded) que doivent subir ces paramètres. [requests] s’en occupe ;

  • lignes 37-40 : le paramètre nommé [auth] est un tuple de deux éléments (login, password). Il représente les identifiants d’une authentification de type Basic ;

• lignes 44-45 : ces deux lignes n’ont qu’un but pédagogique (on les mettra en commentaires lorsque le débogage sera terminé) :

  • [response] représente la réponse HTTP du serveur ;

  • [response.text] représente le texte du document encapsulé dans cette réponse. En phase de débogage, il est utile de vérifier ce que le serveur nous a envoyé ;

• ligne 47 : [response.status_code] est le code de statut HTTP de la réponse reçue. Notre serveur n’en envoie que trois :

  • 200 OK

  • 400 BAD REQUEST

  • 500 INTERNAL SERVER ERROR

• ligne 49 : notre serveur envoie toujours du jSON même en cas d’erreur. La fonction [response.json()] crée un dictionnaire à partir de la chaîne jSON reçue. Rappelons les deux formes possibles pour la chaîne jSON :

   

  Sélectionnez

  1.
  

  {"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"]}}
  

   

  Sélectionnez

  1.
  

  {"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}}}
  

• lignes 51-53 : si le code de statut n’est pas 200 alors on lance une exception avec les messages d’erreur encapsulés dans la réponse ;

• ligne 56 : on récupère le dictionnaire produit par le calcul de l’impôt et on l’utilise pour mettre à jour le paramètre d’entrée [taxpayer] ;

Pour exécuter le client :

• lancer le serveur [server_03] avec le SGBD de votre choix ;

• exécuter le script [main] du client ;

Les résultats seront trouvés dans le dossier [data/output]. Ce sont les mêmes que pour la version 5.