Add static analysis tests for migration anti-patterns

Prevent migration bugs like #64876 (DML before disable_sqlite_fkeys
silently breaking SQLite PRAGMA) from reaching production by adding:

1. AST-based static analysis tests (test_migration_patterns.py) that
   detect three anti-patterns across all migration files:
   - MIG001: DML before disable_sqlite_fkeys (triggers autobegin)
   - MIG002: DDL before disable_sqlite_fkeys (triggers autobegin)
   - MIG003: DML without context.is_offline_mode() guard

2. Unit tests for migration utility functions (test_migration_utils.py)
   covering disable_sqlite_fkeys, mysql_drop_foreignkey_if_exists, and
   ignore_sqlite_value_error with real database backends.

3. Fixes confirmed SQLite PRAGMA bugs in migrations 0100 and 0106 by
   moving all operations inside the disable_sqlite_fkeys block.

4. Adds # noqa: MIG0XX suppression comments (with reasons) to 20 old
   migrations with acknowledged violations, and configures ruff to
   preserve them via external = ["MIG"].

Author: Dev-iL

.pre-commit-config.yaml

@@ -1034,6 +1034,13 @@ repos:
         pass_filenames: false
         files: ^.*\.py$
         require_serial: true
+      - id: check-migration-patterns
+        name: Check migration files for anti-patterns (MIG001/MIG002/MIG003)
+        entry: ./scripts/ci/prek/check_migration_patterns.py
+        language: python
+        pass_filenames: true
+        files: ^airflow-core/src/airflow/migrations/versions/.*\.py$
+        require_serial: true
       ## ADD MOST PREK HOOK ABOVE THAT LINE
       # The below prek hooks are those requiring CI image to be built
       ## ONLY ADD PREK HOOKS HERE THAT REQUIRE CI IMAGE

airflow-core/src/airflow/migrations/versions/0003_2_7_0_add_include_deferred_column_to_pool.py

@@ -43,7 +43,7 @@ def upgrade():
     with op.batch_alter_table("slot_pool") as batch_op:
         batch_op.add_column(sa.Column("include_deferred", sa.Boolean))
     # Different databases support different literal for FALSE. This is fine.
-    op.execute(sa.text(f"UPDATE slot_pool SET include_deferred = {sa.false().compile(op.get_bind())}"))
+    op.execute(sa.text(f"UPDATE slot_pool SET include_deferred = {sa.false().compile(op.get_bind())}"))  # noqa: MIG003 -- backfill default for NOT NULL
     with op.batch_alter_table("slot_pool") as batch_op:
         batch_op.alter_column("include_deferred", existing_type=sa.Boolean, nullable=False)
 

airflow-core/src/airflow/migrations/versions/0027_2_10_3_fix_dag_schedule_dataset_alias_reference_naming.py

@@ -116,7 +116,7 @@ def upgrade():
             ),
             sa.PrimaryKeyConstraint("alias_id", "dag_id", name="dsdar_pkey"),
         )
-        op.execute("INSERT INTO new_table SELECT * FROM dag_schedule_dataset_alias_reference")
+        op.execute("INSERT INTO new_table SELECT * FROM dag_schedule_dataset_alias_reference")  # noqa: MIG003 -- data copy between tables during rename
         op.drop_table("dag_schedule_dataset_alias_reference")
         op.rename_table("new_table", "dag_schedule_dataset_alias_reference")
         op.create_index(
@@ -227,7 +227,7 @@ def downgrade():
             ),
             sa.PrimaryKeyConstraint("alias_id", "dag_id", name="dsdar_pkey"),
         )
-        op.execute("INSERT INTO new_table SELECT * FROM dag_schedule_dataset_alias_reference")
+        op.execute("INSERT INTO new_table SELECT * FROM dag_schedule_dataset_alias_reference")  # noqa: MIG003 -- data copy between tables during rename
         op.drop_table("dag_schedule_dataset_alias_reference")
         op.rename_table("new_table", "dag_schedule_dataset_alias_reference")
         op.create_index(

airflow-core/src/airflow/migrations/versions/0030_3_0_0_rename_schedule_interval_to_timetable_.py

@@ -58,4 +58,4 @@ def downgrade():
             type_=sa.Text,
             nullable=True,
         )
-    op.execute("UPDATE dag SET schedule_interval=NULL;")
+    op.execute("UPDATE dag SET schedule_interval=NULL;")  # noqa: MIG003 -- cleanup during downgrade

airflow-core/src/airflow/migrations/versions/0035_3_0_0_update_user_id_type.py

@@ -57,15 +57,15 @@ def downgrade():
     # Clean up non-numeric user_id values before type conversion to prevent errors
     if dialect_name == "postgresql":
         # For PostgreSQL, use regex to identify non-integer values and set them to NULL
-        op.execute(
+        op.execute(  # noqa: MIG003 -- user ID type reversion during downgrade
             text("""
                 UPDATE dag_run_note
                 SET user_id = NULL
                 WHERE user_id IS NOT NULL
                 AND user_id !~ '^[0-9]+$'
             """)
         )
-        op.execute(
+        op.execute(  # noqa: MIG003 -- user ID type reversion during downgrade
             text("""
                 UPDATE task_instance_note
                 SET user_id = NULL
@@ -75,15 +75,15 @@ def downgrade():
         )
     elif dialect_name == "mysql":
         # For MySQL, use REGEXP to identify non-integer values
-        op.execute(
+        op.execute(  # noqa: MIG003 -- user ID type reversion during downgrade
             text("""
                 UPDATE dag_run_note
                 SET user_id = NULL
                 WHERE user_id IS NOT NULL
                 AND user_id NOT REGEXP '^[0-9]+$'
             """)
         )
-        op.execute(
+        op.execute(  # noqa: MIG003 -- user ID type reversion during downgrade
             text("""
                 UPDATE task_instance_note
                 SET user_id = NULL
@@ -93,15 +93,15 @@ def downgrade():
         )
     elif dialect_name == "sqlite":
         # SQLite doesn't have regex, so use GLOB pattern matching
-        op.execute(
+        op.execute(  # noqa: MIG003 -- user ID type reversion during downgrade
             text("""
                 UPDATE dag_run_note
                 SET user_id = NULL
                 WHERE user_id IS NOT NULL
                 AND (user_id = '' OR user_id GLOB '*[^0-9]*')
             """)
         )
-        op.execute(
+        op.execute(  # noqa: MIG003 -- user ID type reversion during downgrade
             text("""
                 UPDATE task_instance_note
                 SET user_id = NULL

airflow-core/src/airflow/migrations/versions/0036_3_0_0_add_name_field_to_dataset_model.py

@@ -89,7 +89,7 @@ def downgrade():
         # - Join to find if each row has a non-orphaned row with the same URI. If there is, select its id.
         # - Modifying the table currently used for selection is not allowed; the additional layer
         #   SELECT id FROM (...) AS d3 makes MySQL think the selecting table is different.
-        op.execute(
+        op.execute(  # noqa: MIG003 -- data cleanup during downgrade
             "delete from dataset where id in ("
             "    select id from ("
             "        select d1.id from dataset d1"
@@ -100,7 +100,7 @@ def downgrade():
             ")"
         )
     else:
-        op.execute(
+        op.execute(  # noqa: MIG003 -- data cleanup during downgrade
             "delete from dataset as d1 where d1.is_orphaned = true "
             "and exists (select 1 from dataset as d2 where d1.uri = d2.uri and d2.is_orphaned = false)"
         )

airflow-core/src/airflow/migrations/versions/0038_3_0_0_add_asset_active.py

@@ -58,7 +58,7 @@ def upgrade():
         sa.Index("idx_asset_active_name_unique", "name", unique=True),
         sa.Index("idx_asset_active_uri_unique", "uri", unique=True),
     )
-    op.execute("insert into asset_active (name, uri) select name, uri from dataset where is_orphaned = false")
+    op.execute("insert into asset_active (name, uri) select name, uri from dataset where is_orphaned = false")  # noqa: MIG003 -- orphan flag data migration
     with op.batch_alter_table("dataset", schema=None) as batch_op:
         batch_op.drop_column("is_orphaned")
 
@@ -68,7 +68,7 @@ def downgrade():
         batch_op.add_column(
             sa.Column("is_orphaned", sa.Boolean, default=False, nullable=False, server_default="0")
         )
-    op.execute(
+    op.execute(  # noqa: MIG003 -- orphan flag data migration
         "update dataset set is_orphaned = true "
         "where not exists (select 1 from asset_active "
         "where dataset.name = asset_active.name and dataset.uri = asset_active.uri)"

airflow-core/src/airflow/migrations/versions/0042_3_0_0_add_uuid_primary_key_to_task_instance_.py

@@ -240,11 +240,13 @@ def upgrade():
         op.execute(mysql_uuid7_fn)
 
         # Migrate existing rows with UUID v7
-        op.execute("""
+        op.execute(  # noqa: MIG003 -- UUID primary key population
+            """
             UPDATE task_instance
             SET id = uuid_generate_v7(coalesce(queued_dttm, start_date, NOW(3)))
             WHERE id IS NULL
-        """)
+        """
+        )
 
         # Drop this function as it is no longer needed
         op.execute(mysql_uuid7_fn_drop)

airflow-core/src/airflow/migrations/versions/0047_3_0_0_add_dag_versioning.py

@@ -47,8 +47,8 @@ def upgrade():
     """Apply add dag versioning."""
     dialect_name = op.get_bind().dialect.name
 
-    op.execute("delete from dag_code;")
-    op.execute("delete from serialized_dag;")
+    op.execute("delete from dag_code;")  # noqa: MIG003 -- dag versioning data migration
+    op.execute("delete from serialized_dag;")  # noqa: MIG003 -- dag versioning data migration
 
     op.create_table(
         "dag_version",
@@ -160,8 +160,8 @@ def downgrade():
         batch_op.drop_constraint("created_dag_version_id_fkey", type_="foreignkey")
         batch_op.drop_column("created_dag_version_id")
 
-    op.execute("delete from dag_code;")
-    op.execute("delete from serialized_dag;")
+    op.execute("delete from dag_code;")  # noqa: MIG003 -- dag versioning data migration
+    op.execute("delete from serialized_dag;")  # noqa: MIG003 -- dag versioning data migration
 
     with op.batch_alter_table("dag_code", schema=None) as batch_op:
         batch_op.drop_constraint("dag_code_pkey", type_="primary")

airflow-core/src/airflow/migrations/versions/0049_3_0_0_remove_pickled_data_from_xcom_table.py

@@ -243,7 +243,7 @@ def downgrade():
         )
     elif dialect == "mysql":
         op.add_column("xcom", sa.Column("value_blob", LONGBLOB, nullable=True))
-        op.execute("UPDATE xcom SET value_blob = CAST(value AS BINARY);")
+        op.execute("UPDATE xcom SET value_blob = CAST(value AS BINARY);")  # noqa: MIG003 -- type conversion during downgrade
         op.drop_column("xcom", "value")
         op.alter_column("xcom", "value_blob", existing_type=LONGBLOB, new_column_name="value")