testPostgres.py 2.16 KB

import psycopg2
import json
import psycopg2
import json
def recupconf(file):
    f=open(file)
    conf=json.load(f)
    user=conf["identifiant"]["user"]
    password=conf["identifiant"]["password"]
    dbname=conf["dbinfo"]["db"]
    host=conf["dbinfo"]["host"]
    f.close()
    return dbname,host,user,password
class GestionBD:

    #Mise en place et interfaçage d'une base de données Postgres"""

    def __init__(self, dbName , host, user, password):


        try:

            self.baseDonn = psycopg2.connect(dbname=dbName,user =user, host =host, password =password)
        except Exception as err :

            print ('La connexion avec la base de données a echou Erreur détectée :\n%s' % err)

            self.echec =1

        else:

            self.cursor = self.baseDonn.cursor()   # création du curseur
            self.echec =0
    def executerReq(self, req):

        #"Exécution de la requête <req>, avec détection d'erreur éventuelle"

        try:

            self.cursor.execute(req)

        except Exception as err:

            # afficher la requête et le message d'erreur système :

            print ("Requête SQL incorrecte :\n%s\nErreur détectée :\n%s"\

                   % (req, err))

            return 0

        else:

            return 1


    def resultatReq(self):

        #"renvoie le résultat de la requête précédente (un tuple de tuples)"

        resultat=self.cursor.fetchall()
        for row in resultat:
            print(row)
    def listTable(self):

        #"renvoie le résultat de la requête précédente (un tuple de tuples)"
        self.executerReq("select tablename from pg_tables where tablename not like 'pg_%'")
        self.resultatReq()
    def listCTD(self):

        #"renvoie le résultat de la requête précédente (un tuple de tuples)"
        self.executerReq("select count(*) from donnees_ctd")
        self.resultatReq()

    def commit(self):

        if self.baseDonn:

            self.baseDonn.commit()         # transfert curseur -> disque
    def rollback(self):

        if self.baseDonn:

            self.baseDonn.rollback()

    def close(self):

        if self.baseDonn:

            self.baseDonn.close()