refactor: literal parameterized SQL in geofence tag migrations

Replaces psycopg2.sql identifier composition with prebuilt literal queries
(the aux table name and the valid-link filter are compile-time constants),
clearing the odoo-sql-injection-string-format code-scanning alerts. Values
remain parameterized; behavior unchanged (spp_gis suite 92/0).

Author: gonzalesedwin1123

spp_gis/migrations/19.0.2.1.0/post-migration.py

@@ -4,33 +4,30 @@
 Creates one spp.gis.geofence.tag per distinct legacy vocabulary name and
 restores the geofence links, then drops the aux table. See pre-migration.py
 for the full story.
+
+All SQL is literal and value-parameterized; identifiers are never composed.
 """
 
 import logging
 
-from psycopg2 import sql
-
 from odoo import SUPERUSER_ID, api
 
 _logger = logging.getLogger("odoo.addons.spp_gis.migrations.geofence_tags")
 
-AUX_TABLE = "spp_gis_geofence_tag_legacy_migration"
-
 
 def _table_exists(cr, table):
     cr.execute("SELECT 1 FROM information_schema.tables WHERE table_name = %s", (table,))
     return bool(cr.fetchone())
 
 
 def migrate(cr, version):
-    if not _table_exists(cr, AUX_TABLE):
+    if not _table_exists(cr, "spp_gis_geofence_tag_legacy_migration"):
         return
 
     env = api.Environment(cr, SUPERUSER_ID, {})
     Tag = env["spp.gis.geofence.tag"]
-    aux = sql.Identifier(AUX_TABLE)
 
-    cr.execute(sql.SQL("SELECT DISTINCT vocab_name FROM {aux}").format(aux=aux))
+    cr.execute("SELECT DISTINCT vocab_name FROM spp_gis_geofence_tag_legacy_migration")
     tag_ids_by_name = {}
     for (name,) in cr.fetchall():
         tag = Tag.search([("name", "=", name)], limit=1)
@@ -39,7 +36,7 @@ def migrate(cr, version):
         tag_ids_by_name[name] = tag.id
 
     restored = 0
-    cr.execute(sql.SQL("SELECT geofence_id, vocab_name FROM {aux}").format(aux=aux))
+    cr.execute("SELECT geofence_id, vocab_name FROM spp_gis_geofence_tag_legacy_migration")
     for geofence_id, name in cr.fetchall():
         cr.execute(
             """
@@ -51,7 +48,7 @@ def migrate(cr, version):
         )
         restored += cr.rowcount
 
-    cr.execute(sql.SQL("DROP TABLE {aux}").format(aux=aux))
+    cr.execute("DROP TABLE spp_gis_geofence_tag_legacy_migration")
     _logger.info(
         "spp_gis geofence tag migration: created %s tags, restored %s links",
         len(tag_ids_by_name),

spp_gis/migrations/19.0.2.1.0/pre-migration.py

@@ -10,15 +10,55 @@
 This pre-migration moves legacy rows (with the referenced vocabulary's name)
 into an aux table; the matching post-migration recreates them as
 spp.gis.geofence.tag links and drops the aux table.
+
+All SQL is literal and value-parameterized; identifiers are never composed.
 """
 
 import logging
 
-from psycopg2 import sql
-
 _logger = logging.getLogger("odoo.addons.spp_gis.migrations.geofence_tags")
 
-AUX_TABLE = "spp_gis_geofence_tag_legacy_migration"
+_CREATE_AUX = """
+    CREATE TABLE IF NOT EXISTS spp_gis_geofence_tag_legacy_migration (
+        geofence_id integer NOT NULL,
+        vocab_name varchar NOT NULL
+    )
+"""
+
+# name is a translated (jsonb) field: prefer en_US, else any value.
+_PARK_LEGACY = """
+    INSERT INTO spp_gis_geofence_tag_legacy_migration (geofence_id, vocab_name)
+    SELECT rel.geofence_id,
+           COALESCE(v.name->>'en_US', (SELECT t.value FROM jsonb_each_text(v.name) t LIMIT 1))
+    FROM spp_gis_geofence_tag_rel rel
+    JOIN spp_vocabulary v ON v.id = rel.tag_id
+"""
+
+_DELETE_LEGACY = """
+    DELETE FROM spp_gis_geofence_tag_rel rel
+    USING spp_vocabulary v
+    WHERE v.id = rel.tag_id
+"""
+
+# Variants used when the new tag table already exists (migration rerun or
+# tests): a row is a valid new-style link if its tag_id exists there; it is
+# legacy if it references a vocabulary instead. Checking the new table first
+# resolves numeric id collisions in favor of valid links.
+_PARK_LEGACY_SKIP_VALID = """
+    INSERT INTO spp_gis_geofence_tag_legacy_migration (geofence_id, vocab_name)
+    SELECT rel.geofence_id,
+           COALESCE(v.name->>'en_US', (SELECT t.value FROM jsonb_each_text(v.name) t LIMIT 1))
+    FROM spp_gis_geofence_tag_rel rel
+    JOIN spp_vocabulary v ON v.id = rel.tag_id
+    WHERE rel.tag_id NOT IN (SELECT id FROM spp_gis_geofence_tag)
+"""
+
+_DELETE_LEGACY_SKIP_VALID = """
+    DELETE FROM spp_gis_geofence_tag_rel rel
+    USING spp_vocabulary v
+    WHERE v.id = rel.tag_id
+      AND rel.tag_id NOT IN (SELECT id FROM spp_gis_geofence_tag)
+"""
 
 
 def _table_exists(cr, table):
@@ -30,48 +70,17 @@ def migrate(cr, version):
     if not _table_exists(cr, "spp_gis_geofence_tag_rel"):
         return
 
-    # A row is a valid new-style link if its tag_id exists in the new tag
-    # table (only possible when this migration reruns or in tests); it is
-    # legacy if it references a vocabulary instead. Checking the new table
-    # first resolves numeric id collisions in favor of valid links.
     if _table_exists(cr, "spp_gis_geofence_tag"):
-        valid_clause = sql.SQL("AND rel.tag_id NOT IN (SELECT id FROM spp_gis_geofence_tag)")
+        park_query = _PARK_LEGACY_SKIP_VALID
+        delete_query = _DELETE_LEGACY_SKIP_VALID
     else:
-        valid_clause = sql.SQL("")
-
-    aux = sql.Identifier(AUX_TABLE)
-    cr.execute(
-        sql.SQL(
-            """
-            CREATE TABLE IF NOT EXISTS {aux} (
-                geofence_id integer NOT NULL,
-                vocab_name varchar NOT NULL
-            )
-            """
-        ).format(aux=aux)
-    )
-    cr.execute(
-        sql.SQL(
-            """
-            INSERT INTO {aux} (geofence_id, vocab_name)
-            SELECT rel.geofence_id,
-                   COALESCE(v.name->>'en_US', (SELECT t.value FROM jsonb_each_text(v.name) t LIMIT 1))
-            FROM spp_gis_geofence_tag_rel rel
-            JOIN spp_vocabulary v ON v.id = rel.tag_id
-            WHERE true {valid_clause}
-            """
-        ).format(aux=aux, valid_clause=valid_clause)
-    )
+        park_query = _PARK_LEGACY
+        delete_query = _DELETE_LEGACY
+
+    cr.execute(_CREATE_AUX)
+    cr.execute(park_query)
     parked = cr.rowcount
-    cr.execute(
-        sql.SQL(
-            """
-            DELETE FROM spp_gis_geofence_tag_rel rel
-            USING spp_vocabulary v
-            WHERE v.id = rel.tag_id {valid_clause}
-            """
-        ).format(valid_clause=valid_clause)
-    )
+    cr.execute(delete_query)
     _logger.info(
         "spp_gis geofence tag migration: parked %s legacy vocabulary tag links (%s rows removed)",
         parked,