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
| import sys
from pg8000 import DBAPI
from dict_app import *
class GestionBD(object):
"""Mise en place et interfaçage d'une base de données PostgreSQL"""
def __init__(self, dbName, user, passwd, host, port =5432):
"Établissement de la connexion - Création du curseur"
try:
self.baseDonn = DBAPI.connect(host =host, port =port,
database =dbName,
user=user, password=passwd)
except Exception as err:
print('La connexion avec la base de données a échoué :\n'\
'Erreur détectée :\n%s' % err)
self.echec =1
else:
self.cursor = self.baseDonn.cursor() # création du curseur
self.echec =0
def creerTables(self, dicTables):
"Création des tables décrites dans le dictionnaire <dicTables>."
for table in dicTables: # parcours des clés du dictionnaire
req = "CREATE TABLE %s (" % table
pk =''
for descr in dicTables[table]:
nomChamp = descr[0]
tch = descr[1]
if tch =='i':
typeChamp ='INTEGER'
elif tch =='k':
# champ 'clé primaire' (entier incrémenté automatiquement)
typeChamp ='SERIAL'
pk = nomChamp
else:
typeChamp ='VARCHAR(%s)' % tch
req = req + "%s %s, " % (nomChamp, typeChamp)
if pk == '':
req = req[:-2] + ")"
else:
req = req + "CONSTRAINT %s_pk PRIMARY KEY(%s))" % (pk, pk)
self.executerReq(req)
def supprimerTables(self, dicTables):
"Suppression de toutes les tables décrites dans <dicTables>"
for table in list(dicTables.keys()):
req ="DROP TABLE %s" % table
self.executerReq(req)
self.commit() # transfert -> disque
def executerReq(self, req, param =None):
"Exécution de la requête <req>, avec détection d'erreur éventuelle"
try:
self.cursor.execute(req, param)
except Exception as err:
# afficher la requête et le message d'erreur système :
print("Requête SQL incorrecte :\n{}\nErreur détectée :".format(req))
print(err)
return 0
else:
return 1
def resultatReq(self):
"renvoie le résultat de la requête précédente (une liste de tuples)"
return self.cursor.fetchall()
def commit(self):
if self.baseDonn:
self.baseDonn.commit() # transfert curseur -> disque
def close(self):
if self.baseDonn:
self.baseDonn.close() |
Partager