Add a bundle for example dags when enabled (apache#45533)

Once we start parsing from bundles, we will have a separate bundle to
represent the example dags, instead of simply adding them to the list of
files from the dags folder like we do today.

airflow/dag_processing/bundles/manager.py

@@ -64,6 +64,23 @@ def parse_config(self) -> None:
                 "Bundle config is not a list. Check config value"
                 " for section `dag_bundles` and key `backends`."
             )
+
+        # example dags!
+        if conf.getboolean("core", "LOAD_EXAMPLES"):
+            from airflow import example_dags
+
+            example_dag_folder = next(iter(example_dags.__path__))
+            backends.append(
+                {
+                    "name": "example_dags",
+                    "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
+                    "kwargs": {
+                        "local_folder": example_dag_folder,
+                        "refresh_interval": conf.getint("scheduler", "dag_dir_list_interval"),
+                    },
+                }
+            )
+
         seen = set()
         for cfg in backends:
             name = cfg["name"]

tests/dag_processing/bundles/test_dag_bundle_manager.py

@@ -68,7 +68,9 @@
 )
 def test_parse_bundle_config(value, expected):
     """Test that bundle_configs are read from configuration."""
-    envs = {"AIRFLOW__DAG_BUNDLES__BACKENDS": value} if value else {}
+    envs = {"AIRFLOW__CORE__LOAD_EXAMPLES": "False"}
+    if value:
+        envs["AIRFLOW__DAG_BUNDLES__BACKENDS"] = value
     cm = nullcontext()
     exp_fail = False
     if isinstance(expected, str):
@@ -133,6 +135,7 @@ def clear_db():
 
 
 @pytest.mark.db_test
+@conf_vars({("core", "LOAD_EXAMPLES"): "False"})
 def test_sync_bundles_to_db(clear_db):
     def _get_bundle_names_and_active():
         with create_session() as session:
@@ -167,3 +170,14 @@ def test_view_url(version):
     with patch.object(BaseDagBundle, "view_url") as view_url_mock:
         bundle_manager.view_url("my-test-bundle", version=version)
     view_url_mock.assert_called_once_with(version=version)
+
+
+def test_example_dags_bundle_added():
+    manager = DagBundlesManager()
+    manager.parse_config()
+    assert "example_dags" in manager._bundle_config
+
+    with conf_vars({("core", "LOAD_EXAMPLES"): "False"}):
+        manager = DagBundlesManager()
+        manager.parse_config()
+        assert "example_dags" not in manager._bundle_config