Move config item dag_bundle_storage_path to dag_bundles section (#45697)

airflow/config_templates/config.yml

@@ -28,14 +28,6 @@ core:
       type: string
       example: ~
       default: "{AIRFLOW_HOME}/dags"
-    dag_bundle_storage_path:
-      description: |
-        The folder where Airflow bundles can store files locally (if required).
-        By default, this is ``tempfile.gettempdir()/airflow``. This path must be absolute.
-      version_added: 3.0.0
-      type: string
-      example: "`tempfile.gettempdir()/dag_bundles"
-      default: ~
     hostname_callable:
       description: |
         Hostname by providing a path to a callable, which will resolve the hostname.
@@ -2670,7 +2662,17 @@ dag_bundles:
     Configuration for the DAG bundles. This allows Airflow to load DAGs from different sources.
 
   options:
-    backends:
+    dag_bundle_storage_path:
+      description: |
+        String path to folder where Airflow bundles can store files locally. Not templated.
+        If no path is provided, Airflow will use ``Path(tempfile.gettempdir()) / "airflow"``.
+        This path must be absolute.
+      version_added: 3.0.0
+      type: string
+      example: "/tmp/some-place"
+      default: ~
+
+    config_list:
       description: |
         List of backend configs.  Must supply name, classpath, and kwargs for each backend.
 

airflow/dag_processing/bundles/base.py

@@ -74,7 +74,7 @@ def _dag_bundle_root_storage_path(self) -> Path:
 
         This is the root path, shared by various bundles. Each bundle should have its own subdirectory.
         """
-        if configured_location := conf.get("core", "dag_bundle_storage_path", fallback=None):
+        if configured_location := conf.get("dag_bundles", "dag_bundle_storage_path", fallback=None):
             return Path(configured_location)
         return Path(tempfile.gettempdir(), "airflow", "dag_bundles")
 

airflow/dag_processing/bundles/manager.py

@@ -54,7 +54,7 @@ def parse_config(self) -> None:
         if self._bundle_config:
             return
 
-        backends = conf.getjson("dag_bundles", "backends")
+        backends = conf.getjson("dag_bundles", "config_list")
 
         if not backends:
             return

providers/tests/fab/auth_manager/conftest.py

@@ -95,7 +95,7 @@ def _config_bundle(path_to_parse: Path | str):
                 "kwargs": {"path": str(path_to_parse), "refresh_interval": 0},
             }
         ]
-        with conf_vars({("dag_bundles", "backends"): json.dumps(bundle_config)}):
+        with conf_vars({("dag_bundles", "config_list"): json.dumps(bundle_config)}):
             yield
 
     return _config_bundle

task_sdk/tests/execution_time/test_supervisor.py

@@ -74,7 +74,7 @@ def lineno():
 
 def local_dag_bundle_cfg(path, name="my-bundle"):
     return {
-        "AIRFLOW__DAG_BUNDLES__BACKENDS": json.dumps(
+        "AIRFLOW__DAG_BUNDLES__CONFIG_LIST": json.dumps(
             [
                 {
                     "name": name,

task_sdk/tests/execution_time/test_task_runner.py

@@ -130,7 +130,7 @@ def test_parse(test_dags_dir: Path, make_ti_context):
     with patch.dict(
         os.environ,
         {
-            "AIRFLOW__DAG_BUNDLES__BACKENDS": json.dumps(
+            "AIRFLOW__DAG_BUNDLES__CONFIG_LIST": json.dumps(
                 [
                     {
                         "name": "my-bundle",
@@ -574,7 +574,7 @@ def test_dag_parsing_context(make_ti_context, mock_supervisor_comms, monkeypatch
         ]
     )
 
-    monkeypatch.setenv("AIRFLOW__DAG_BUNDLES__BACKENDS", dag_bundle_val)
+    monkeypatch.setenv("AIRFLOW__DAG_BUNDLES__CONFIG_LIST", dag_bundle_val)
     ti, _ = startup()
 
     # Presence of `conditional_task` below means DAG ID is properly set in the parsing context!

tests/conftest.py

@@ -111,7 +111,7 @@ def _config_bundle(path_to_parse: Path | str):
                 "kwargs": {"path": str(path_to_parse), "refresh_interval": 0},
             }
         ]
-        with conf_vars({("dag_bundles", "backends"): json.dumps(bundle_config)}):
+        with conf_vars({("dag_bundles", "config_list"): json.dumps(bundle_config)}):
             yield
 
     return _config_bundle

tests/dag_processing/bundles/test_dag_bundle_manager.py

@@ -70,7 +70,7 @@ def test_parse_bundle_config(value, expected):
     """Test that bundle_configs are read from configuration."""
     envs = {"AIRFLOW__CORE__LOAD_EXAMPLES": "False"}
     if value:
-        envs["AIRFLOW__DAG_BUNDLES__BACKENDS"] = value
+        envs["AIRFLOW__DAG_BUNDLES__CONFIG_LIST"] = value
     cm = nullcontext()
     exp_fail = False
     if isinstance(expected, str):
@@ -108,7 +108,7 @@ def path(self):
 def test_get_bundle():
     """Test that get_bundle builds and returns a bundle."""
 
-    with patch.dict(os.environ, {"AIRFLOW__DAG_BUNDLES__BACKENDS": json.dumps(BASIC_BUNDLE_CONFIG)}):
+    with patch.dict(os.environ, {"AIRFLOW__DAG_BUNDLES__CONFIG_LIST": json.dumps(BASIC_BUNDLE_CONFIG)}):
         bundle_manager = DagBundlesManager()
 
         with pytest.raises(ValueError, match="'bundle-that-doesn't-exist' is not configured"):
@@ -120,7 +120,7 @@ def test_get_bundle():
     assert bundle.refresh_interval == 1
 
     # And none for version also works!
-    with patch.dict(os.environ, {"AIRFLOW__DAG_BUNDLES__BACKENDS": json.dumps(BASIC_BUNDLE_CONFIG)}):
+    with patch.dict(os.environ, {"AIRFLOW__DAG_BUNDLES__CONFIG_LIST": json.dumps(BASIC_BUNDLE_CONFIG)}):
         bundle = bundle_manager.get_bundle(name="my-test-bundle")
     assert isinstance(bundle, BasicBundle)
     assert bundle.name == "my-test-bundle"
@@ -144,7 +144,7 @@ def _get_bundle_names_and_active():
             )
 
     # Initial add
-    with patch.dict(os.environ, {"AIRFLOW__DAG_BUNDLES__BACKENDS": json.dumps(BASIC_BUNDLE_CONFIG)}):
+    with patch.dict(os.environ, {"AIRFLOW__DAG_BUNDLES__CONFIG_LIST": json.dumps(BASIC_BUNDLE_CONFIG)}):
         manager = DagBundlesManager()
         manager.sync_bundles_to_db()
     assert _get_bundle_names_and_active() == [("my-test-bundle", True)]
@@ -156,13 +156,13 @@ def _get_bundle_names_and_active():
     assert _get_bundle_names_and_active() == [("dags-folder", True), ("my-test-bundle", False)]
 
     # Re-enable one that reappears in config
-    with patch.dict(os.environ, {"AIRFLOW__DAG_BUNDLES__BACKENDS": json.dumps(BASIC_BUNDLE_CONFIG)}):
+    with patch.dict(os.environ, {"AIRFLOW__DAG_BUNDLES__CONFIG_LIST": json.dumps(BASIC_BUNDLE_CONFIG)}):
         manager = DagBundlesManager()
         manager.sync_bundles_to_db()
     assert _get_bundle_names_and_active() == [("dags-folder", False), ("my-test-bundle", True)]
 
 
-@conf_vars({("dag_bundles", "backends"): json.dumps(BASIC_BUNDLE_CONFIG)})
+@conf_vars({("dag_bundles", "config_list"): json.dumps(BASIC_BUNDLE_CONFIG)})
 @pytest.mark.parametrize("version", [None, "hello"])
 def test_view_url(version):
     """Test that view_url calls the bundle's view_url method."""
@@ -185,6 +185,6 @@ def test_example_dags_bundle_added():
 
 def test_example_dags_name_is_reserved():
     reserved_name_config = [{"name": "example_dags"}]
-    with conf_vars({("dag_bundles", "backends"): json.dumps(reserved_name_config)}):
+    with conf_vars({("dag_bundles", "config_list"): json.dumps(reserved_name_config)}):
         with pytest.raises(AirflowConfigException, match="Bundle name 'example_dags' is a reserved name."):
             DagBundlesManager().parse_config()

tests/dag_processing/test_dag_bundles.py

@@ -39,12 +39,12 @@
 
 @pytest.fixture(autouse=True)
 def bundle_temp_dir(tmp_path):
-    with conf_vars({("core", "dag_bundle_storage_path"): str(tmp_path)}):
+    with conf_vars({("dag_bundles", "dag_bundle_storage_path"): str(tmp_path)}):
         yield tmp_path
 
 
 def test_default_dag_storage_path():
-    with conf_vars({("core", "dag_bundle_storage_path"): ""}):
+    with conf_vars({("dag_bundles", "dag_bundle_storage_path"): ""}):
         bundle = LocalDagBundle(name="test", path="/hello")
         assert bundle._dag_bundle_root_storage_path == Path(tempfile.gettempdir(), "airflow", "dag_bundles")
 
@@ -60,7 +60,7 @@ def get_current_version(self):
         def path(self):
             pass
 
-    with conf_vars({("core", "dag_bundle_storage_path"): None}):
+    with conf_vars({("dag_bundles", "dag_bundle_storage_path"): None}):
         bundle = BasicBundle(name="test")
         assert bundle._dag_bundle_root_storage_path == Path(tempfile.gettempdir(), "airflow", "dag_bundles")
 

tests/dag_processing/test_manager.py

@@ -857,7 +857,7 @@ def test_bundles_are_refreshed(self):
         bundletwo.refresh_interval = 300
         bundletwo.get_current_version.return_value = None
 
-        with conf_vars({("dag_bundles", "backends"): json.dumps(config)}):
+        with conf_vars({("dag_bundles", "config_list"): json.dumps(config)}):
             DagBundlesManager().sync_bundles_to_db()
             with mock.patch(
                 "airflow.dag_processing.bundles.manager.DagBundlesManager"
@@ -910,7 +910,7 @@ def test_bundles_versions_are_stored(self):
         mybundle.supports_versioning = True
         mybundle.get_current_version.return_value = "123"
 
-        with conf_vars({("dag_bundles", "backends"): json.dumps(config)}):
+        with conf_vars({("dag_bundles", "config_list"): json.dumps(config)}):
             DagBundlesManager().sync_bundles_to_db()
             with mock.patch(
                 "airflow.dag_processing.bundles.manager.DagBundlesManager"