"""
pyFormation.gestionPostgreSql
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Le module implémente la gestion de la base PostgreSql.
:copyright: (c) 2015 par moulinux.
:license: GPLv3, voir la LICENCE pour plus d'informations.
"""
import psycopg2
import psycopg2.extras
import config
[docs]class GestionBd:
"""Mise en place et interfaçage d'une base de données PostgreSQL"""
[docs] def __init__(self):
"""Établissement de la connexion - Création du curseur"""
try:
self.baseDonn = \
psycopg2.connect(dbname=config.DB_NAME,
user=config.USER, password=config.PASSWD,
host=config.HOST)
except psycopg2.DatabaseError as err:
print('La connexion avec la base de données a échoué :\n',
'Erreur détectée :{}'.format(err))
self.echec = True
else:
# création d'un curseur de type dictionnaire :
self.baseDonn.autocommit = True
self.cursor = \
self.baseDonn.cursor(cursor_factory=psycopg2.extras.DictCursor)
self.echec = False
[docs] def executerReq(self, req, params):
"""Exécution de la requête <req>, avec détection d'erreur éventuelle"""
try:
self.cursor.execute(req, params)
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, err))
return False
else:
return True
[docs] def resultatReq(self):
"""
renvoie le résultat de la requête précédente (une liste de tuples)
"""
return self.cursor.fetchall()
[docs] def valider(self):
"""transfert du curseur vers le disque"""
if self.baseDonn:
self.baseDonn.commit()
[docs] def fermer(self):
"""fermeture du curseur"""
if self.baseDonn:
self.cursor.close()
self.baseDonn.close()
[docs] def connBase(self):
"""
indique si la connexion a réussie ou échouée.
:return: True si la connexion a échouée, False sinon
:rtype: booléen
"""
return self.echec
[docs] def obtenirActivites(self):
"""
Retourne le jeu des enregistrements des activités.
:return: jeu des enregistrements des activités
:rtype: list
"""
reqSql = """SELECT * FROM "Activite"
ORDER BY numero"""
if self.executerReq(reqSql, ()):
return self.resultatReq()
return None
[docs] def obtenirActions(self, numActivite):
"""
Retourne le jeu des enregistrements des actions de formation
correspondant à une activité.
:param numActivite: activité correspondant aux actions de formation
:type numActivite: entier
:return: jeu des enregistrements des actions de formation
:rtype: list
"""
reqSql = """SELECT * FROM "Action"
WHERE "numeroActivite"=%s
ORDER BY code"""
if self.executerReq(reqSql, (numActivite,)):
return self.resultatReq()
return None
[docs] def obtenirSessions(self, uneAction):
"""
Retourne le jeu des enregistrements des sessions
correspondant à une action de formation.
:param uneAction: action correspondant aux sessions
:type uneAction: chaîne de caractères
:return: jeu des enregistrements des sessions
:rtype: list
"""
reqSql = """SELECT S.*, L.libelle AS "nomLieu"
FROM "Session" AS S, "Lieu" AS L
WHERE "codeAction"=%s
AND S."idLieu" = L."id"
ORDER BY numero"""
if self.executerReq(reqSql, (uneAction,)):
return self.resultatReq()
return None
[docs] def obtenirParametres(self):
"""
Retourne le jeu des enregistrements des paramètres.
:return: enregistrement des paramètres
:rtype: dictionnaire
"""
reqSql = """SELECT * FROM "Parametre"
ORDER BY id"""
if self.executerReq(reqSql, ()):
return self.resultatReq()[0]
return None
[docs] def obtenirLieux(self):
"""
Retourne le jeu des enregistrements des lieux.
:return: jeu des enregistrements des lieux
:rtype: list
"""
reqSql = """SELECT * FROM "Lieu"
ORDER BY id"""
if self.executerReq(reqSql, ()):
return self.resultatReq()
return None
[docs] def detailSession(self, numSession):
"""
Détaille une session.
:param numSession: identifiant de la session
:type numSession: entier
:return: enregistrement de la session
:rtype: dictionnaire
"""
reqSql = """SELECT * FROM "Session"
WHERE numero=%s"""
if self.executerReq(reqSql, (numSession,)):
return self.resultatReq()[0]
return None
[docs] def creerSession(self, unTupSession):
"""
Crée une nouvelle session.
:param unTupSession: paramètres de la session
:type unTupSession: tuple
:return: True si la création a réussie, False sinon
:rtype: booléen
"""
reqSql = """INSERT INTO "Session" ("codeAction", "idLieu",
"nombreMaxParticipant", "dateSession", "coutPrevu",
"forfaitJournalier", "distanceMin")
VALUES (%s,%s,%s,%s,%s,%s,%s)"""
test = self.executerReq(reqSql, unTupSession)
return test
[docs] def modifierSession(self, unTupSession):
"""
Modifie une session.
:param unTupSession: paramètres de la session
:type unTupSession: tuple
:return: True si la modification a réussie, False sinon
:rtype: booléen
"""
reqSql = """UPDATE "Session" SET "idLieu"=%s,
"nombreMaxParticipant"=%s, "dateSession"=%s,
"coutPrevu"=%s, "coutReel"=%s
WHERE numero=%s"""
test = self.executerReq(reqSql, unTupSession)
return test
[docs] def supprimerSession(self, numSession):
"""
Suppprime une session.
:param numSession: identifiant de la session
:type numSession: entier
:return: True si la suppression a réussie, False sinon
:rtype: booléen
"""
reqSql = """DELETE FROM "Session" WHERE "numero"=%s"""
test = self.executerReq(reqSql, (numSession,))
return test
[docs] def trouverNbInscrits(self, numSession):
"""
Trouve le nombre d'agent inscrits à une session.
:param numSession: identifiant de la session
:type numSession: entier
:return: le nombre d'inscrits
:rtype: entier
"""
reqSql = """SELECT COUNT(*) as "nbInscrits"
FROM "Inscription"
WHERE "numeroSession"=%s"""
if self.executerReq(reqSql, (numSession,)):
return self.resultatReq()[0]['nbInscrits']
return 0
