Disable foreign keys on sqlite when modifying dag_run (#22848)

If we do not disable FKs then it has the side effect of deleting all task instances.

GitOrigin-RevId: d328dcfd0a94a9018fea7a7248a799469e24356c

airflow/migrations/utils.py

@@ -16,6 +16,7 @@
 # under the License.
 
 from collections import defaultdict
+from contextlib import contextmanager
 
 
 def get_mssql_table_constraints(conn, table_name):
@@ -41,3 +42,13 @@ def get_mssql_table_constraints(conn, table_name):
     for constraint, constraint_type, col_name in result:
         constraint_dict[constraint_type][constraint].append(col_name)
     return constraint_dict
+
+
+@contextmanager
+def disable_sqlite_fkeys(op):
+    if op.get_bind().dialect.name == 'sqlite':
+        op.execute("PRAGMA foreign_keys=off")
+        yield op
+        op.execute("PRAGMA foreign_keys=on")
+    else:
+        yield op

airflow/migrations/versions/0099_f9da662e7089_add_task_log_filename_template_model.py

@@ -26,6 +26,7 @@
 from alembic import op
 from sqlalchemy import Column, ForeignKey, Integer, Text
 
+from airflow.migrations.utils import disable_sqlite_fkeys
 from airflow.utils.sqlalchemy import UtcDateTime
 
 # Revision identifiers, used by Alembic.
@@ -50,7 +51,8 @@ def upgrade():
         Integer,
         ForeignKey("log_template.id", name="task_instance_log_template_id_fkey", ondelete="NO ACTION"),
     )
-    with op.batch_alter_table("dag_run") as batch_op:
+
+    with disable_sqlite_fkeys(op), op.batch_alter_table("dag_run") as batch_op:
         batch_op.add_column(dag_run_log_filename_id)