Managing Feature Deletions and Triggers in SQLite with Python Integration (QGIS)

Question

I am working with Spatialite layers on QGIS for a project that is also used by colleagues who are unfamiliar with SQLite, meaning they won’t be running SQLite queries themselves.

The issue I’m facing is that SQLite cannot differentiate between occasional deletions and a complete deletion of the table’s content. Each deletion is processed independently, regardless of the context.

Using SQLite on a Spatialite DB doesn’t allow me to use transactions (or at least, I haven’t been able to use them effectively). One specific problem arises when deleting several features from table A, which is related to tables B, C, and D.

If I delete 25 features from table A, SQLite will execute all the dependent triggers on the other tables. For the first 24 deletions, this process is unnecessary, but the 25th deletion requires the trigger to run. This results in redundant operations, slowing down the process.

I couldn’t find a solution using just SQLite, so I thought about using a Python script to check the number of selected features on the map and save this value in an SQLite table.

Since I’m not experienced with Python, I wasn’t sure of the best approach to program this solution.

After some research and help from "AI", I managed to come up with the following code. It works, but I’m unsure if anything is missing or if there’s a more efficient way to implement it. Also, it stops working whenever I restart QGIS.

import os

from qgis.core import QgsProject from PyQt5.QtCore import QTimer import sqlite3

project_path = QgsProject.instance().fileName() project_dir = os.path.dirname(project_path) db_path = os.path.join(project_dir, "ECOMED-JBI.sqlite")

layer_name = "ESPECES_PT" layer = QgsProject.instance().mapLayersByName(layer_name)

if layer: layer = layer[0] update_timer = QTimer() update_timer.setSingleShot(True)

def on_selection_changed():
 update_timer.start(300) # Attente de 300 ms avant mise à jour
def update_db():
 count = layer.selectedFeatureCount()
 print(f"Sélection: {count} entités")
 try:
 with sqlite3.connect(db_path) as conn:
 cursor = conn.cursor()
 cursor.execute("UPDATE table_selection SET name_lay = ?, count_sel = ? WHERE id = 1", 
 (layer_name, count))
 conn.commit()
 except sqlite3.Error as e:
 print(f"Erreur SQLite : {e}")
update_timer.timeout.connect(update_db)
layer.selectionChanged.connect(on_selection_changed)
print(f"Écouteur activé pour {layer_name}")

Since everything is done from the map (no SQLite queries), saving the number of selected features in an SQLite table would allow me to better control what is executed and when.

This issue is related to the one here: Link to issue on GIS stack exchange Where I've been advised to post here.

Answer 1

The best solution I found is to activate macros during starting, with this code:

 from qgis.core import QgsProject
from qgis.utils import iface
from PyQt5.QtCore import QTimer
import sqlite3
import os
# Variables globales pour la macro
timer = None
tracked_layers = {}
db_path = None
# Définition des correspondances couche -> ID
LAYER_IDS = {
 "ESPECES_PT": 1,
 "HABITATS_POLY": 2,
 "ZONES_HUMIDES_POLY": 3
}
def openProject():
 global timer, tracked_layers, db_path
 
 # Initialisation du timer
 timer = QTimer()
 timer.setSingleShot(True)
 
 # Récupération du chemin de la base de données
 project_path = QgsProject.instance().absolutePath()
 db_path = os.path.join(project_path, 'ECOMED-JBI.sqlite')
 
 def on_selection_changed(layer_name):
 timer.start(300) # Attente de 300 ms avant mise à jour
 
 def update_db():
 try:
 with sqlite3.connect(db_path) as conn:
 cursor = conn.cursor()
 for layer_name, layer in tracked_layers.items():
 count = layer.selectedFeatureCount()
 layer_id = LAYER_IDS[layer_name]
 print(f"Mise à jour {layer_name} (ID={layer_id}): {count} entités")
 
 cursor.execute("""
 UPDATE table_selection 
 SET count_sel = ? 
 WHERE id = ?
 """, (count, layer_id))
 
 conn.commit()
 iface.messageBar().pushMessage(
 "Succès", 
 "Mise à jour des sélections effectuée", 
 level=3, 
 duration=2
 )
 except sqlite3.Error as e:
 print(f"Erreur SQLite : {e}")
 iface.messageBar().pushMessage(
 "Erreur", 
 f"Erreur lors de la mise à jour de la base : {str(e)}", 
 level=2, 
 duration=5
 )
 
 # Connexion du timer à la fonction de mise à jour
 timer.timeout.connect(update_db)
 
 # Initialisation du suivi pour chaque couche
 project = QgsProject.instance()
 for layer_name in LAYER_IDS.keys():
 try:
 layer = project.mapLayersByName(layer_name)[0]
 tracked_layers[layer_name] = layer
 
 def make_handler(name):
 return lambda: on_selection_changed(name)
 
 layer.selectionChanged.connect(make_handler(layer_name))
 print(f"Écouteur activé pour {layer_name}")
 
 except IndexError:
 print(f"Attention: La couche {layer_name} n'a pas été trouvée")
 iface.messageBar().pushMessage(
 "Attention", 
 f"La couche {layer_name} n'a pas été trouvée", 
 level=1, 
 duration=5
 )
 
def saveProject():
 pass
def closeProject():
 global timer, tracked_layers
 if timer:
 timer.stop()
 
 for layer in tracked_layers.values():
 try:
 layer.selectionChanged.disconnect()
 except:
 pass
 
 tracked_layers.clear()