Move dags-folder bundle to LocalDagBundle; rename kwarg (#45721)

This moves the dags-folder bundle to use LocalDagBundle, instead of
having its own class. A follow up will remove the `[core] dags_folder`
config as well, but that will be much larger reaching.

I've also renamed the `local_folder` kwarg to `path` - just a nicer
name for it.

airflow/config_templates/config.yml

@@ -2707,7 +2707,7 @@ dag_bundles:
         [
             {{
               "name": "dags-folder",
-              "classpath": "airflow.dag_processing.bundles.dagfolder.DagsFolderDagBundle",
+              "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
               "kwargs": {{}}
             }}
             ]

airflow/dag_processing/bundles/local.py

@@ -19,21 +19,25 @@
 
 from pathlib import Path
 
+from airflow import settings
 from airflow.dag_processing.bundles.base import BaseDagBundle
 
 
 class LocalDagBundle(BaseDagBundle):
     """
     Local DAG bundle - exposes a local directory as a DAG bundle.
 
-    :param local_folder: Local folder where the DAGs are stored
+    :param path: Local path where the DAGs are stored
     """
 
     supports_versioning = False
 
-    def __init__(self, *, local_folder: str, **kwargs) -> None:
+    def __init__(self, *, path: str | None = None, **kwargs) -> None:
         super().__init__(**kwargs)
-        self._path = Path(local_folder)
+        if path is None:
+            path = settings.DAGS_FOLDER
+
+        self._path = Path(path)
 
     def get_current_version(self) -> None:
         return None

airflow/dag_processing/bundles/manager.py

@@ -81,7 +81,7 @@ def parse_config(self) -> None:
                     "name": "example_dags",
                     "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
                     "kwargs": {
-                        "local_folder": example_dag_folder,
+                        "path": example_dag_folder,
                     },
                 }
             )

providers/tests/fab/auth_manager/conftest.py

@@ -92,7 +92,7 @@ def _config_bundle(path_to_parse: Path | str):
             {
                 "name": "testing",
                 "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
-                "kwargs": {"local_folder": str(path_to_parse), "refresh_interval": 0},
+                "kwargs": {"path": str(path_to_parse), "refresh_interval": 0},
             }
         ]
         with conf_vars({("dag_bundles", "backends"): json.dumps(bundle_config)}):

task_sdk/tests/execution_time/test_supervisor.py

@@ -79,7 +79,7 @@ def local_dag_bundle_cfg(path, name="my-bundle"):
                 {
                     "name": name,
                     "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
-                    "kwargs": {"local_folder": str(path), "refresh_interval": 1},
+                    "kwargs": {"path": str(path), "refresh_interval": 1},
                 }
             ]
         )

task_sdk/tests/execution_time/test_task_runner.py

@@ -135,7 +135,7 @@ def test_parse(test_dags_dir: Path, make_ti_context):
                     {
                         "name": "my-bundle",
                         "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
-                        "kwargs": {"local_folder": str(test_dags_dir), "refresh_interval": 1},
+                        "kwargs": {"path": str(test_dags_dir), "refresh_interval": 1},
                     }
                 ]
             ),
@@ -569,7 +569,7 @@ def test_dag_parsing_context(make_ti_context, mock_supervisor_comms, monkeypatch
             {
                 "name": "my-bundle",
                 "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
-                "kwargs": {"local_folder": str(test_dags_dir), "refresh_interval": 1},
+                "kwargs": {"path": str(test_dags_dir), "refresh_interval": 1},
             }
         ]
     )

tests/conftest.py

@@ -108,7 +108,7 @@ def _config_bundle(path_to_parse: Path | str):
             {
                 "name": "testing",
                 "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
-                "kwargs": {"local_folder": str(path_to_parse), "refresh_interval": 0},
+                "kwargs": {"path": str(path_to_parse), "refresh_interval": 0},
             }
         ]
         with conf_vars({("dag_bundles", "backends"): json.dumps(bundle_config)}):

tests/dag_processing/bundles/test_dag_bundle_manager.py

@@ -50,7 +50,7 @@
                     {
                         "name": "my-bundle",
                         "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
-                        "kwargs": {"local_folder": "/tmp/hihi", "refresh_interval": 1},
+                        "kwargs": {"path": "/tmp/hihi", "refresh_interval": 1},
                     }
                 ]
             ),

tests/dag_processing/test_dag_bundles.py

@@ -25,7 +25,6 @@
 from git import Repo
 
 from airflow.dag_processing.bundles.base import BaseDagBundle
-from airflow.dag_processing.bundles.dagfolder import DagsFolderDagBundle
 from airflow.dag_processing.bundles.git import GitDagBundle, GitHook
 from airflow.dag_processing.bundles.local import LocalDagBundle
 from airflow.exceptions import AirflowException
@@ -46,7 +45,7 @@ def bundle_temp_dir(tmp_path):
 
 def test_default_dag_storage_path():
     with conf_vars({("core", "dag_bundle_storage_path"): ""}):
-        bundle = LocalDagBundle(name="test", local_folder="/hello")
+        bundle = LocalDagBundle(name="test", path="/hello")
         assert bundle._dag_bundle_root_storage_path == Path(tempfile.gettempdir(), "airflow", "dag_bundles")
 
 
@@ -68,24 +67,22 @@ def path(self):
 
 class TestLocalDagBundle:
     def test_path(self):
-        bundle = LocalDagBundle(name="test", local_folder="/hello")
+        bundle = LocalDagBundle(name="test", path="/hello")
         assert bundle.path == Path("/hello")
 
+    @conf_vars({("core", "dags_folder"): "/tmp/somewhere/dags"})
+    def test_path_default(self):
+        bundle = LocalDagBundle(name="test", refresh_interval=300)
+        assert bundle.path == Path("/tmp/somewhere/dags")
+
     def test_none_for_version(self):
         assert LocalDagBundle.supports_versioning is False
 
-        bundle = LocalDagBundle(name="test", local_folder="/hello")
+        bundle = LocalDagBundle(name="test", path="/hello")
 
         assert bundle.get_current_version() is None
 
 
-class TestDagsFolderDagBundle:
-    @conf_vars({("core", "dags_folder"): "/tmp/somewhere/dags"})
-    def test_path(self):
-        bundle = DagsFolderDagBundle(name="test")
-        assert bundle.path == Path("/tmp/somewhere/dags")
-
-
 GIT_DEFAULT_BRANCH = "main"
 
 

tests/dag_processing/test_manager.py

@@ -839,12 +839,12 @@ def test_bundles_are_refreshed(self):
             {
                 "name": "bundleone",
                 "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
-                "kwargs": {"local_folder": "/dev/null", "refresh_interval": 0},
+                "kwargs": {"path": "/dev/null", "refresh_interval": 0},
             },
             {
                 "name": "bundletwo",
                 "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
-                "kwargs": {"local_folder": "/dev/null", "refresh_interval": 300},
+                "kwargs": {"path": "/dev/null", "refresh_interval": 300},
             },
         ]
 
@@ -900,7 +900,7 @@ def test_bundles_versions_are_stored(self):
             {
                 "name": "mybundle",
                 "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
-                "kwargs": {"local_folder": "/dev/null", "refresh_interval": 0},
+                "kwargs": {"path": "/dev/null", "refresh_interval": 0},
             },
         ]