XI. Exercice [IMPOTS] avec MySQL▲
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.
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 :
La connexion à la base [dbimpots] se fera avec l'identité (root,"").
Nous allons utiliser l'architecture suivante :
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.
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.
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 :
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 :
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.
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.
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.
