You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ka...@apache.org on 2025/04/30 09:52:39 UTC
(airflow) 02/04: Auto-correct stale bundle_name in DAG versions when DAG bundle configuration changes (#49886)
This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v3-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit de981ccc9ba8bf04ff28403601a36555ad7af8e2
Author: Rahul Vats <43...@users.noreply.github.com>
AuthorDate: Wed Apr 30 14:51:28 2025 +0530
Auto-correct stale bundle_name in DAG versions when DAG bundle configuration changes (#49886)
Auto-correct stale bundle_name in DAG versions when DAG bundle configuration changes (#49886)
(cherry picked from commit c675d742943f13d11b553aab2c14809fdec17f4d)
---
.../src/airflow/dag_processing/bundles/manager.py | 24 ++++++++++++++--
.../bundles/test_dag_bundle_manager.py | 33 ++++++++++++++++++----
2 files changed, 50 insertions(+), 7 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/bundles/manager.py b/airflow-core/src/airflow/dag_processing/bundles/manager.py
index 18e855230bc..9760da617a1 100644
--- a/airflow-core/src/airflow/dag_processing/bundles/manager.py
+++ b/airflow-core/src/airflow/dag_processing/bundles/manager.py
@@ -20,6 +20,7 @@ from typing import TYPE_CHECKING
from airflow.configuration import conf
from airflow.exceptions import AirflowConfigException
+from airflow.models.dag_version import DagVersion
from airflow.models.dagbundle import DagBundleModel
from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.module_loading import import_string
@@ -123,17 +124,36 @@ class DagBundlesManager(LoggingMixin):
def sync_bundles_to_db(self, *, session: Session = NEW_SESSION) -> None:
self.log.debug("Syncing DAG bundles to the database")
stored = {b.name: b for b in session.query(DagBundleModel).all()}
- for name in self._bundle_config.keys():
+ active_bundle_names = set(self._bundle_config.keys())
+ for name in active_bundle_names:
if bundle := stored.pop(name, None):
bundle.active = True
else:
session.add(DagBundleModel(name=name))
self.log.info("Added new DAG bundle %s to the database", name)
-
+ inactive_bundle_names = []
for name, bundle in stored.items():
bundle.active = False
+ inactive_bundle_names.append(name)
self.log.warning("DAG bundle %s is no longer found in config and has been disabled", name)
+ if inactive_bundle_names and active_bundle_names:
+ new_bundle_name = sorted(active_bundle_names)[0]
+ updated_rows = (
+ session.query(DagVersion)
+ .filter(DagVersion.bundle_name.in_(inactive_bundle_names))
+ .update(
+ {DagVersion.bundle_name: new_bundle_name},
+ synchronize_session=False,
+ )
+ )
+
+ self.log.info(
+ "Updated %d DAG versions from inactive bundles to active bundle %s",
+ updated_rows,
+ new_bundle_name,
+ )
+
def get_bundle(self, name: str, version: str | None = None) -> BaseDagBundle:
"""
Get a DAG bundle by name.
diff --git a/airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py b/airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py
index febb687690b..f61f6ff4896 100644
--- a/airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py
+++ b/airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py
@@ -27,7 +27,9 @@ import pytest
from airflow.dag_processing.bundles.base import BaseDagBundle
from airflow.dag_processing.bundles.manager import DagBundlesManager
from airflow.exceptions import AirflowConfigException
+from airflow.models.dag_version import DagVersion
from airflow.models.dagbundle import DagBundleModel
+from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.utils.session import create_session
from tests_common.test_utils.config import conf_vars
@@ -139,13 +141,17 @@ def clear_db():
@pytest.mark.db_test
@conf_vars({("core", "LOAD_EXAMPLES"): "False"})
-def test_sync_bundles_to_db(clear_db):
+def test_sync_bundles_to_db(clear_db, dag_maker):
def _get_bundle_names_and_active():
with create_session() as session:
return (
session.query(DagBundleModel.name, DagBundleModel.active).order_by(DagBundleModel.name).all()
)
+ def _get_dag_version_bundle_names():
+ with create_session() as session:
+ return session.query(DagVersion.dag_id, DagVersion.bundle_name).all()
+
# Initial add
with patch.dict(
os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(BASIC_BUNDLE_CONFIG)}
@@ -154,11 +160,23 @@ def test_sync_bundles_to_db(clear_db):
manager.sync_bundles_to_db()
assert _get_bundle_names_and_active() == [("my-test-bundle", True)]
- # simulate bundle config change
- # note: airflow will detect config changes when they are in env vars
+ # Create DAG version with 'my-test-bundle'
+ with dag_maker(dag_id="test_dag", schedule=None):
+ EmptyOperator(task_id="mytask")
+ with create_session() as session:
+ session.add(DagVersion(dag_id="test_dag", version_number=1, bundle_name="my-test-bundle"))
+
+ # simulate bundle config change (now 'dags-folder' is active, 'my-test-bundle' becomes inactive)
manager = DagBundlesManager()
manager.sync_bundles_to_db()
- assert _get_bundle_names_and_active() == [("dags-folder", True), ("my-test-bundle", False)]
+ assert _get_bundle_names_and_active() == [
+ ("dag_maker", False),
+ ("dags-folder", True),
+ ("my-test-bundle", False),
+ ]
+
+ # Check that the DAG version bundle_name got auto-updated to active one
+ assert _get_dag_version_bundle_names() == [("test_dag", "dags-folder")]
# Re-enable one that reappears in config
with patch.dict(
@@ -166,7 +184,12 @@ def test_sync_bundles_to_db(clear_db):
):
manager = DagBundlesManager()
manager.sync_bundles_to_db()
- assert _get_bundle_names_and_active() == [("dags-folder", False), ("my-test-bundle", True)]
+ assert _get_bundle_names_and_active() == [
+ ("dag_maker", False),
+ ("dags-folder", False),
+ ("my-test-bundle", True),
+ ]
+ assert _get_dag_version_bundle_names() == [("test_dag", "my-test-bundle")]
@conf_vars({("dag_processor", "dag_bundle_config_list"): json.dumps(BASIC_BUNDLE_CONFIG)})