Option authentification pour table avec droits particuliers.

This commit is contained in:
Tom LAVEILLE 2026-05-22 15:05:31 +02:00
parent 5fd101543e
commit e23b9e80c0

View File

@ -28,6 +28,7 @@ from qgis.core import (
from qgis.PyQt.QtWidgets import (
QDialog,
QPushButton,
QInputDialog,
QTableWidgetItem,
QMessageBox,
QVBoxLayout,
@ -65,13 +66,16 @@ itemIconVecteur.setIcon(icon)
# Récupération des informations de connexion à la base de données
try:
account = login_base('account')
user = account[0] # Nom d'utilisateur
mdp = account[1] # Mot de passe
host = account[2] # Hôte de la base de données
port = account[3] # Port de connexion
dbname = account[4] # Nom de la base de données
sigdb = account[5] # Base de données SIG
refdb = account[6] # Base de données REF
user = account[0] # Nom d'utilisateur
user_save = account[0] # Nom d'utilisateur sauvegarde
mdp = account[1] # Mot de passe
mdp_save = account[1] # Mot de passe sauvegard
host = account[2] # Hôte de la base de données
port = account[3] # Port de connexion
dbname = account[4] # Nom de la base de données
sigdb = account[5] # Base de données SIG
refdb = account[6] # Base de données REF
except NameError:
print('Fails to login DB for account')
@ -90,6 +94,7 @@ class Flux_Editor(QDialog, EDITOR_CLASS):
Classe principale de l'éditeur de flux
Gère l'interface de sélection et de chargement des couches depuis les bases de données
"""
def __init__(self, parent=None):
"""Initialisation de l'interface et des composants"""
_ = parent
@ -178,6 +183,10 @@ class Flux_Editor(QDialog, EDITOR_CLASS):
self.DeBUG.hide()
def bd_source(self):
global user, mdp
if user != user_save or mdp != mdp_save:
user = user_save
mdp = mdp_save
"""Sélectionne la base de données source (SIG ou REF)"""
self.activateWindow()
bd_origine = self.comboBox_2.currentText()
@ -573,6 +582,7 @@ class Flux_Editor(QDialog, EDITOR_CLASS):
self.chargement_flux()
def chargement_flux(self):
global cur, con, dbtype
"""Charge les couches sélectionnées dans QGIS"""
managerAU = QgsApplication.authManager()
managerAU.availableAuthMethodConfigs().keys()
@ -608,9 +618,36 @@ class Flux_Editor(QDialog, EDITOR_CLASS):
for row in range(0, self.tableWidget_2.rowCount()):
color_rgba_db = 855030089 # Code couleur pour couche dans autre BD
color_rgba_droit = 851150528 # Code couleur pour droits insuffisants
self.QMBquestionNoLogin = QMessageBox()
if self.tableWidget_2.item(row, 1).background().color().rgba() == color_rgba_droit:
self.QMBquestion = QMessageBox.question(iface.mainWindow(), u"Attention !", "Vous ne disposez pas des droit pour la couche «" + str(self.tableWidget_2.item(row, 1).text()) + ' ' + str(self.tableWidget_2.item(row, 2).text()) + "» !", QMessageBox.StandardButton(0x00004000))
elif self.tableWidget_2.item(row, 1).background().color().rgba() != color_rgba_db:
self.QMBquestionNoLogin.setWindowTitle(u"Attention !")
self.QMBquestionNoLogin.setIcon(QMessageBox.Icon.Warning)
self.QMBquestionNoLogin.setText("Vous ne disposez pas des droit pour la couche «" + str(self.tableWidget_2.item(row, 1).text()) + ' ' + str(self.tableWidget_2.item(row, 2).text()) + "» !")
self.QMBquestionNoLogin.setStandardButtons(QMessageBox.StandardButton(0x00004000) | QMessageBox.StandardButton(0x00080000))
self.QMBquestionNoLogin = self.QMBquestionNoLogin.exec()
# self.QMBquestionNoLogin = QMessageBox.question(iface.mainWindow(), u"Attention !", "Vous ne disposez pas des droit pour la couche «" + str(self.tableWidget_2.item(row, 1).text()) + ' ' + str(self.tableWidget_2.item(row, 2).text()) + "» !", QMessageBox.StandardButton(0x00004000), QMessageBox.StandardButton(0x00080000))
if self.QMBquestionNoLogin != QMessageBox.StandardButton(0x00004000):
self.InputDialog = QInputDialog()
user_bypass = (self.InputDialog.getText(None, 'Foncier', 'Identifiant:'))[0]
mdp_bypass = (self.InputDialog.getText(None, 'Foncier', 'Mot de pass:'))[0]
global user, mdp
user = user_bypass
mdp = mdp_bypass
try:
con = psycopg2.connect("host=" + host + " port=" + port + " dbname=" + dbtype + " user=" + user + " password=" + mdp)
cur = con.cursor(cursor_factory=psycopg2.extras.DictCursor)
except psycopg2.OperationalError:
print('password authentication failed')
user = user_save
mdp = mdp_save
else:
if user != user_save or mdp != mdp_save:
user = user_save
mdp = mdp_save
if (self.tableWidget_2.item(row, 1).background().color().rgba() != color_rgba_db) and (self.QMBquestionNoLogin != QMessageBox.StandardButton(0x00004000)):
# supression de la partie de l'url après le point d'interrogation
if dbtype == sigdb:
code = self.tableWidget_2.item(row, 1).text()
@ -652,33 +689,42 @@ class Flux_Editor(QDialog, EDITOR_CLASS):
# Détection du type de géométrie
geom_type = 'SELECT right(st_geometrytype(geom),-3) as a FROM ' + schema + '.' + table + ' GROUP BY a'
cur.execute(geom_type)
list_typegeom = cur.fetchall()
if len(list_typegeom) > 1:
for typegeom in list_typegeom:
if typegeom[0] == 'MultiPolygon':
uri.setWkbType(QgsWkbTypes.MultiPolygon)
elif typegeom[0] == 'MultiLineString':
uri.setWkbType(QgsWkbTypes.MultiLineString)
elif typegeom[0] == 'Point':
uri.setWkbType(QgsWkbTypes.Point)
if typegeom[0] == 'Polygon':
uri.setWkbType(QgsWkbTypes.Polygon)
elif typegeom[0] == 'LineString':
uri.setWkbType(QgsWkbTypes.LineString)
elif typegeom[0] == 'MultiPoint':
uri.setWkbType(QgsWkbTypes.MultiPoint)
if (typegeom[0] is not None and typegeom[0] != 'Polygon'):
uri.setSrid('2154')
layer = QgsVectorLayer(uri.uri(), table, "postgres")
# Ajout de la couche au canevas QGIS
QgsProject.instance().addMapLayer(layer)
else:
layer = QgsVectorLayer(uri.uri(), table, "postgres")
# Ajout de la couche au canevas QGIS
QgsProject.instance().addMapLayer(layer)
try:
cur.execute(geom_type)
list_typegeom = cur.fetchall()
if len(list_typegeom) > 1:
for typegeom in list_typegeom:
if typegeom[0] == 'MultiPolygon':
uri.setWkbType(QgsWkbTypes.MultiPolygon)
elif typegeom[0] == 'MultiLineString':
uri.setWkbType(QgsWkbTypes.MultiLineString)
elif typegeom[0] == 'Point':
uri.setWkbType(QgsWkbTypes.Point)
if typegeom[0] == 'Polygon':
uri.setWkbType(QgsWkbTypes.Polygon)
elif typegeom[0] == 'LineString':
uri.setWkbType(QgsWkbTypes.LineString)
elif typegeom[0] == 'MultiPoint':
uri.setWkbType(QgsWkbTypes.MultiPoint)
if (typegeom[0] is not None and typegeom[0] != 'Polygon'):
uri.setSrid('2154')
layer = QgsVectorLayer(uri.uri(), table, "postgres")
# Ajout de la couche au canevas QGIS
QgsProject.instance().addMapLayer(layer)
else:
layer = QgsVectorLayer(uri.uri(), table, "postgres")
# Ajout de la couche au canevas QGIS
QgsProject.instance().addMapLayer(layer)
except psycopg2.errors.UndefinedTable:
print('Table not exist... RollBack !')
con.rollback()
except psycopg2.errors.InsufficientPrivilege:
print('Insufficient Privilege... RollBack !')
con.rollback()
else:
self.QMBquestion = QMessageBox.question(iface.mainWindow(), u"Attention !", "La couche «" + str(self.tableWidget_2.item(row, 1).text()) + ' ' + str(self.tableWidget_2.item(row, 2).text()) + "» ne ce trouve pas dans cette BD !", QMessageBox.StandardButton(0x00004000))
if self.QMBquestionNoLogin != QMessageBox.StandardButton(0x00004000):
self.QMBquestion = QMessageBox.question(iface.mainWindow(), u"Attention !", "La couche «" + str(self.tableWidget_2.item(row, 1).text()) + ' ' + str(self.tableWidget_2.item(row, 2).text()) + "» ne ce trouve pas dans cette BD !", QMessageBox.StandardButton(0x00004000))
def combobox_custom(self):
"""Remplit la liste déroulante des catégories selon la base de données sélectionnée"""