Slightly faster downgrade tests

.github/workflows/ci.yml

@@ -1064,17 +1064,17 @@ ${{ hashFiles('.pre-commit-config.yaml') }}"
       - name: "Test downgrade migration file ${{ env.BACKEND }}"
         run: >
           breeze shell "airflow db reset --skip-init -y &&
-          airflow db upgrade --to-revision heads &&
+          airflow db upgrade --to-revision heads --no-reserialize-dags &&
           airflow db downgrade -r e959f08ac86c -y &&
-          airflow db upgrade"
+          airflow db upgrade --no-reserialize-dags"
         env:
           IMAGE_TAG: ${{ env.IMAGE_TAG_FOR_THE_BUILD }}
       - name: "Test downgrade ORM ${{ env.BACKEND }}"
         run: >
           breeze shell "airflow db reset -y &&
-          airflow db upgrade &&
+          airflow db upgrade --no-reserialize-dags &&
           airflow db downgrade -r e959f08ac86c -y &&
-          airflow db upgrade"
+          airflow db upgrade --no-reserialize-dags"
         env:
           IMAGE_TAG: ${{ env.IMAGE_TAG_FOR_THE_BUILD }}
       - name: "Test Offline SQL generation ${{ env.BACKEND }}"
@@ -1156,17 +1156,17 @@ ${{ hashFiles('.pre-commit-config.yaml') }}"
       - name: "Test downgrade migration file ${{ env.BACKEND }}"
         run: >
           breeze shell "airflow db reset --skip-init -y &&
-          airflow db upgrade --to-revision heads &&
+          airflow db upgrade --to-revision heads --no-reserialize-dags &&
           airflow db downgrade -r e959f08ac86c -y &&
-          airflow db upgrade"
+          airflow db upgrade --no-reserialize-dags"
         env:
           IMAGE_TAG: ${{ env.IMAGE_TAG_FOR_THE_BUILD }}
       - name: "Test downgrade ORM ${{ env.BACKEND }}"
         run: >
           breeze shell "airflow db reset -y &&
-          airflow db upgrade &&
+          airflow db upgrade --no-reserialize-dags &&
           airflow db downgrade -r e959f08ac86c -y &&
-          airflow db upgrade"
+          airflow db upgrade --no-reserialize-dags"
         env:
           IMAGE_TAG: ${{ env.IMAGE_TAG_FOR_THE_BUILD }}
       - name: "Tests: ${{needs.build-info.outputs.test-types}}"

airflow/cli/cli_parser.py

@@ -547,6 +547,14 @@ def string_lower_type(val):
     type=int,
     default=60,
 )
+ARG_DB_RESERIALIZE_DAGS = Arg(
+    ("--no-reserialize-dags",),
+    # Not intended for user, so dont show in help
+    help=argparse.SUPPRESS,
+    action="store_false",
+    default=True,
+    dest="reserialize_dags",
+)
 ARG_DB_VERSION__UPGRADE = Arg(
     ("-n", "--to-version"),
     help=(
@@ -1445,6 +1453,7 @@ class GroupCommand(NamedTuple):
             ARG_DB_SQL_ONLY,
             ARG_DB_FROM_REVISION,
             ARG_DB_FROM_VERSION,
+            ARG_DB_RESERIALIZE_DAGS,
         ),
     ),
     ActionCommand(

airflow/cli/commands/db_command.py

@@ -81,7 +81,12 @@ def upgradedb(args):
     else:
         print("Generating sql for upgrade -- upgrade commands will *not* be submitted.")
 
-    db.upgradedb(to_revision=to_revision, from_revision=from_revision, show_sql_only=args.show_sql_only)
+    db.upgradedb(
+        to_revision=to_revision,
+        from_revision=from_revision,
+        show_sql_only=args.show_sql_only,
+        reserialize_dags=args.reserialize_dags,
+    )
     if not args.show_sql_only:
         print("Upgrades done")
 

airflow/migrations/env.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import contextlib
+import sys
 from logging.config import fileConfig
 
 from alembic import context
@@ -114,3 +115,9 @@ def run_migrations_online():
     run_migrations_offline()
 else:
     run_migrations_online()
+
+if 'airflow.www.app' in sys.modules:
+    # Already imported, make sure we clear out any cached app
+    from airflow.www.app import purge_cached_app
+
+    purge_cached_app()

airflow/migrations/versions/0074_2_0_0_resource_based_permissions.py

@@ -27,7 +27,7 @@
 import logging
 
 from airflow.security import permissions
-from airflow.www.app import create_app
+from airflow.www.app import cached_app
 
 # revision identifiers, used by Alembic.
 revision = '2c6edca13270'
@@ -288,7 +288,7 @@
 
 def remap_permissions():
     """Apply Map Airflow permissions."""
-    appbuilder = create_app(config={'FAB_UPDATE_PERMS': False}).appbuilder
+    appbuilder = cached_app(config={'FAB_UPDATE_PERMS': False}).appbuilder
     for old, new in mapping.items():
         (old_resource_name, old_action_name) = old
         old_permission = appbuilder.sm.get_permission(old_action_name, old_resource_name)
@@ -313,7 +313,7 @@ def remap_permissions():
 
 def undo_remap_permissions():
     """Unapply Map Airflow permissions"""
-    appbuilder = create_app(config={'FAB_UPDATE_PERMS': False}).appbuilder
+    appbuilder = cached_app(config={'FAB_UPDATE_PERMS': False}).appbuilder
     for old, new in mapping.items():
         (new_resource_name, new_action_name) = new[0]
         new_permission = appbuilder.sm.get_permission(new_action_name, new_resource_name)

airflow/migrations/versions/0078_2_0_1_remove_can_read_permission_on_config_.py

@@ -27,7 +27,7 @@
 import logging
 
 from airflow.security import permissions
-from airflow.www.app import create_app
+from airflow.www.app import cached_app
 
 # revision identifiers, used by Alembic.
 revision = '82b7c48c147f'
@@ -42,7 +42,7 @@ def upgrade():
     log = logging.getLogger()
     handlers = log.handlers[:]
 
-    appbuilder = create_app(config={'FAB_UPDATE_PERMS': False}).appbuilder
+    appbuilder = cached_app(config={'FAB_UPDATE_PERMS': False}).appbuilder
     roles_to_modify = [role for role in appbuilder.sm.get_all_roles() if role.name in ["User", "Viewer"]]
     can_read_on_config_perm = appbuilder.sm.get_permission(
         permissions.ACTION_CAN_READ, permissions.RESOURCE_CONFIG
@@ -59,7 +59,7 @@ def upgrade():
 
 def downgrade():
     """Add can_read action on config resource for User and Viewer role"""
-    appbuilder = create_app(config={'FAB_UPDATE_PERMS': False}).appbuilder
+    appbuilder = cached_app(config={'FAB_UPDATE_PERMS': False}).appbuilder
     roles_to_modify = [role for role in appbuilder.sm.get_all_roles() if role.name in ["User", "Viewer"]]
     can_read_on_config_perm = appbuilder.sm.get_permission(
         permissions.ACTION_CAN_READ, permissions.RESOURCE_CONFIG

airflow/migrations/versions/0084_2_1_0_resource_based_permissions_for_default_.py

@@ -27,7 +27,7 @@
 import logging
 
 from airflow.security import permissions
-from airflow.www.app import create_app
+from airflow.www.app import cached_app
 
 # revision identifiers, used by Alembic.
 revision = 'a13f7613ad25'
@@ -140,7 +140,7 @@
 
 def remap_permissions():
     """Apply Map Airflow permissions."""
-    appbuilder = create_app(config={'FAB_UPDATE_PERMS': False}).appbuilder
+    appbuilder = cached_app(config={'FAB_UPDATE_PERMS': False}).appbuilder
     for old, new in mapping.items():
         (old_resource_name, old_action_name) = old
         old_permission = appbuilder.sm.get_permission(old_action_name, old_resource_name)
@@ -165,7 +165,7 @@ def remap_permissions():
 
 def undo_remap_permissions():
     """Unapply Map Airflow permissions"""
-    appbuilder = create_app(config={'FAB_UPDATE_PERMS': False}).appbuilder
+    appbuilder = cached_app(config={'FAB_UPDATE_PERMS': False}).appbuilder
     for old, new in mapping.items():
         (new_resource_name, new_action_name) = new[0]
         new_permission = appbuilder.sm.get_permission(new_action_name, new_resource_name)

airflow/utils/db.py

@@ -825,8 +825,7 @@ def check_and_run_migrations():
         sys.exit(1)
 
 
-@provide_session
-def reserialize_dags(*, session: Session = NEW_SESSION) -> None:
+def _reserialize_dags(*, session: Session) -> None:
     from airflow.models.dagbag import DagBag
     from airflow.models.serialized_dag import SerializedDagModel
 
@@ -1476,6 +1475,7 @@ def upgradedb(
     to_revision: str | None = None,
     from_revision: str | None = None,
     show_sql_only: bool = False,
+    reserialize_dags: bool = True,
     session: Session = NEW_SESSION,
 ):
     """
@@ -1556,7 +1556,8 @@ def upgradedb(
                 os.environ['AIRFLOW__DATABASE__SQL_ALCHEMY_MAX_SIZE'] = val
             settings.reconfigure_orm()
 
-    reserialize_dags(session=session)
+    if reserialize_dags:
+        _reserialize_dags(session=session)
     add_default_pool_if_not_exists(session=session)
     synchronize_log_template(session=session)
 

docs/apache-airflow/img/airflow_erd.sha256

@@ -1 +1 @@
-dca9bf08dd97b5f51c387726a2e9d25996769e85dda0d403e8e088ce222faa09
+3f6af8d3dbaf44cbef18f738e568b9a66375cc1214d6da9e482ebe5c591d24dc

tests/cli/commands/test_db_command.py

@@ -96,7 +96,7 @@ def test_cli_check_migrations(self, mock_wait_for_migrations):
     @mock.patch("airflow.cli.commands.db_command.db.upgradedb")
     def test_cli_upgrade_success(self, mock_upgradedb, args, called_with):
         db_command.upgradedb(self.parser.parse_args(['db', 'upgrade', *args]))
-        mock_upgradedb.assert_called_once_with(**called_with)
+        mock_upgradedb.assert_called_once_with(**called_with, reserialize_dags=True)
 
     @pytest.mark.parametrize(
         'args, pattern',