Introduction au langage Python par l'exemple


précédentsommairesuivant

XI. Exercice [IMPOTS] avec MySQL

Image non disponible

XI-A. Transfert d'un fichier texte dans une table MySQL

Le script à venir va transférer les données du fichier texte suivant :

 
Sélectionnez
1.
2.
3.
12620:13190:15640:24740:31810:39970:48360:55790:92970:127860:151250:172040:195000:0
0:0.05:0.1:0.15:0.2:0.25:0.3:0.35:0.4:0.45:0.5:0.55:0.6:0.65
0:631:1290.5:272.5:3309.5:4900:6898.5:9316.5:12106:16754.5:23147.5:30710:39312:49062

dans la table [impots] de la base MySQL [dbimpots] suivante :

Image non disponible

La connexion à la base [dbimpots] se fera avec l'identité (root,"").

Nous allons utiliser l'architecture suivante :

Image non disponible

Le script [console] que nous allons écrire va utiliser la classe [ImpotsFile] pour accéder aux données du fichier texte. L'accès en écriture à la base de données se fera avec les méthodes vues précédemment.

Le code du script est le suivant :

Programme (impotstxt2mysql)
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.
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.
104.
105.
106.
107.
108.
109.
110.
111.
112.
113.
114.
115.
116.
117.
118.
119.
120.
# -*- coding=utf-8 -*-

# import du module des classes Impots
from impots import *
import re

# --------------------------------------------------------------------------
def copyToMysql(limites,coeffR,coeffN,HOTE,USER,PWD,BASE,TABLE):
    # copie les 3 tableaux numériques limites, coeffR, coeffN
    # dans la table TABLE de la base mysql BASE
    # la base mysql est sur la machine HOTE
    # l'utilisateur est identifié par USER et PWD

    # on met les requêtes SQL dans une liste
    # suppression de la table
    requetes=["drop table %s" % (TABLE)]
    # création de la table
    requete="create table %s (limites decimal(10,2), coeffR decimal(6,2), coeffN decimal(10,2))" % (TABLE)
    requetes.append(requete)
    # remplissage
    for i in range(len(limites)):
        # requête d'insertion
        requetes.append("insert into %s (limites,coeffR,coeffN) values (%s,%s,%s)" % (TABLE,limites[i],coeffR[i],coeffN[i]))
    # on exécute les ordres SQL
    return executerCommandes(HOTE,USER,PWD,BASE,requetes,False,False)
    

def executerCommandes(HOTE,ID,PWD,BASE,requetes,suivi=False,arret=True):
    # utilise la connexion (HOTE,ID,PWD,BASE)
    # exécute sur cette connexion les commandes SQL contenues dans la liste requetes
    # si suivi=True alors chaque exécution d'un ordre SQL fait l'objet d'un affichage indiquant sa réussite ou son échec
    # si arret=False, la fonction s'arrête sur la 1re erreur rencontrée sinon elle exécute toutes les commandes SQL
    # la fonction rend une liste (nb d'erreurs, erreur1, erreur2...)

    # connexion
    try:
        connexion=MySQLdb.connect(host=HOTE,user=ID,passwd=PWD,db=BASE)
    except MySQLdb.OperationalError,erreur:
        return [1,"Erreur lors de la connexion a MySql sous l'identite (%s,%s,%s,%s) : %s" % (HOTE, ID, PWD, BASE, erreur)]
      
    # on demande un curseur
    curseur=connexion.cursor()
    # exécution des requêtes SQL contenues dans la liste requetes
    # on les exécute - au départ pas d'erreur
    erreurs=[0]
    for i in range(len(requetes)):
        # on met la requête dans une variable locale
        requete=requetes[i]
        # a-t-on une requête vide ?
        if re.match(r"^\s*$",requete):
            continue
        # exécution de la requête i
        erreur=""
        try:
            curseur.execute(requete)
        except Exception, erreur:
            pass
        # y a-t-il eu une erreur ?
        if erreur:
            # une erreur de plus
            erreurs[0]+=1
            # msg d'erreur
            msg="%s : Erreur (%s)" % (requete[0:len(requete)-1],erreur)
            erreurs.append(msg)
            # suivi écran ou non ?
            if suivi:
                print msg
            # on s'arrête ?
            if arret:
                return erreurs
        else:
            if suivi: 
                # on affiche la requête sans sa marque de fin de ligne
                print "%s : Execution reussie" % (cutNewLineChar(requete))
                # infos sur le résultat de la requête exécutée
                afficherInfos(curseur)
    # fermeture connexion
    try:
        connexion.commit()
        connexion.close()
    except MySQLdb.OperationalError,erreur:
        # une erreur de plus
        erreurs[0]+=1
        # msg d'erreur
        msg="%s : Erreur (%s)" % (requete,erreur)
        erreurs.append(msg)

    # retour
    return erreurs


# ------------------------------------------------ main
# l'identité de l'utilisateur
USER="root"
PWD=""
# la machine hôte du SGBD
HOTE="localhost"
# identité de la base
BASE="dbimpots"
# identité de la table des données
TABLE="impots"
# le fichier des données
IMPOTS="impots.txt"

# instanciation couche [dao]
try:
    dao=ImpotsFile(IMPOTS)
except (IOError, ImpotsError) as infos:
    print ("Une erreur s'est produite : {0}".format(infos))
    sys.exit()

# on transfère les données récupérées dans une table mysql
erreurs=copyToMysql(dao.limites,dao.coeffR,dao.coeffN,HOTE,USER,PWD,BASE,TABLE)
if erreurs[0]:
    for i in range(1,len(erreurs)):
        print erreurs[i]
else:
    print "Transfert opere"
# fin
sys.exit()

Notes :

  • lignes 106-110 : on instancie la classe [ImpotsFile] présentée page  ;
  • ligne 113 : les tableaux limites, coeffR, coeffN sont transférés dans une base MySQL ;
  • ligne 8 : la fonction copyToMysql exécute ce transfert. La fonction copyToMysql crée un tableau des requêtes à exécuter et les fait exécuter par la fonction executerCommandes, ligne 25 ;
  • lignes 28-89 : la fonction executerCommandes est celle déjà présentée précédemment avec une différence : au lieu d'être dans un fichier texte, les requêtes sont dans une liste ;
Le fichier texte impots.txt
Sélectionnez
1.
2.
3.
12620:13190:15640:24740:31810:39970:48360:55790:92970:127860:151250:172040:195000:0
0:0.05:0.1:0.15:0.2:0.25:0.3:0.35:0.4:0.45:0.5:0.55:0.6:0.65
0:631:1290.5:272.5:3309.5:4900:6898.5:9316.5:12106:16754.5:23147.5:30710:39312:49062

Les résultats écran :

Transfert opéré

Vérification avec phpMyAdmin :

Image non disponible

XI-B. Le programme de calcul de l'impôt

Maintenant que les données nécessaires au calcul de l'impôt sont dans une base de données, nous pouvons écrire le script de calcul de l'impôt. Nous utilisons de nouveau une architecture trois couches :

Image non disponible

La nouvelle couche [DAO] sera connectée au SGBD MySQL et sera implémentée par la classe [ImpotsMySQL]. Elle offrira à la couche [metier] la même interface que précédemment constituée de l'unique méthode getData qui rend le tuple (limites, coeffR, coeffN). Ainsi la couche [metier] ne changera pas vis-à-vis de la version précédente.

XI-C. La classe [ImpotsMySQL]

La couche [DAO] est maintenant implémentée par la classe [ImpotsMySQL] suivante (fichier impots.py)  :

ImpotsMySQL
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.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
class ImpotsMySQL:

    # constructeur
    def __init__(self,HOTE,USER,PWD,BASE,TABLE):
        # initialise les attributs limites, coeffR, coeffN
        # les données nécessaires au calcul de l'impôt ont été placées dans la table mysqL TABLE
        # appartenant à la base BASE. La table a la structure suivante
        # limites decimal(10,2), coeffR decimal(10,2), coeffN decimal(10,2)
        # la connexion à la base mysql de la machine HOTE se fait sous l'identité (USER,PWD)
        # lance une exception si éventuelle erreur
        
        # connexion à la base mysql
        connexion=MySQLdb.connect(host=HOTE,user=USER,passwd=PWD,db=BASE)
        # on demande un curseur
        curseur=connexion.cursor()
        
        # lecture en bloc de la table TABLE
        requete="select limites,coeffR,coeffN from %s" % (TABLE)
        # exécute la requête [requete] sur la base [base] de la connexion [connexion]
        curseur.execute(requete)
        # exploitation du résultat de la requête
        ligne=curseur.fetchone()
        self.limites=[]
        self.coeffR=[]
        self.coeffN=[]
        while(ligne):
            # ligne courante
            self.limites.append(ligne[0])
            self.coeffR.append(ligne[1])
            self.coeffN.append(ligne[2])
            # ligne suivante
            ligne=curseur.fetchone()
        # déconnexion
        connexion.close()

    def getData(self):
        return (self.limites, self.coeffR, self.coeffN)

Notes :

  • ligne 18 : la requête SQL SELECT qui demande les données de la base de données MySql. Les lignes résultant du SELECT sont ensuite exploitées une par une par [curseur.fetchone] (lignes 22 et 32) pour créer les tableaux limites, coeffR, coeffN (lignes 28-30) ;
  • la méthode getData de l'interface de la couche [DAO].

XI-D. Le script console

Le code du script console (impots_04) est le suivant :

script console (impots_04)
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.
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.
# -*- coding=utf-8 -*-

# import du module des classes Impots*
from impots import *

# ------------------------------------------------ main
# l'identité de l'utilisateur
USER="root"
PWD=""
# la machine hôte du sgbd
HOTE="localhost"
# identité de la base
BASE="dbimpots"
# identité de la table des données
TABLE="impots"
# fichier d'entrées
DATA="data.txt"
# fichier de sorties
RESULTATS="resultats.txt"

# instanciation couche [metier]
try:
    metier=ImpotsMetier(ImpotsMySQL(HOTE,USER,PWD,BASE,TABLE))
except (IOError, ImpotsError) as infos:
    print ("Une erreur s'est produite : {0}".format(infos))
    sys.exit()

# les données nécessaires au calcul de l'impôt ont été placées dans le fichier IMPOTS
# à raison d'une ligne par tableau sous la forme
# val1:val2:val3...

# lecture des données
try:
    data=open(DATA,"r")
except:
    print "Impossible d'ouvrir en lecture le fichier des donnees [DATA]"
    sys.exit()

# ouverture fichier des résultats
try:
    resultats=open(RESULTATS,"w")
except:  
    print "Impossible de creer le fichier des résultats [RESULTATS]"
    sys.exit()

# utilitaires
u=Utilitaires()

# on exploite la ligne courante du fichier des données
ligne=data.readline()
while(ligne != ''):
    # on enlève l'éventuelle marque de fin de ligne
    ligne=u.cutNewLineChar(ligne)
    # on récupère les 3 champs marié:enfants:salaire qui forment la ligne
    (marie,enfants,salaire)=ligne.split(",")
    enfants=int(enfants)
    salaire=int(salaire)
    # on calcule l'impôt
    impot=metier.calculer(marie,enfants,salaire)
    # on inscrit le résultat
    resultats.write("{0}:{1}:{2}:{3}\n".format(marie,enfants,salaire,impot))
    # on lit une nouvelle ligne
    ligne=data.readline()
# on ferme les fichiers
data.close()
resultats.close()

Notes :

  • ligne 23 : instanciation des couches [DAO] et [metier] ;
  • le reste du code est connu.

Résultats

Les mêmes qu'avec les versions précédentes de l'exercice.


précédentsommairesuivant

  

Licence Creative Commons
Le contenu de cet article est rédigé par Serge Tahé et est mis à disposition selon les termes de la Licence Creative Commons Attribution - Pas d’Utilisation Commerciale - Partage dans les Mêmes Conditions 3.0 non transposé.
Les logos Developpez.com, en-tête, pied de page, css, et look & feel de l'article sont Copyright © 2013 Developpez.com.