Restore DagBag include_examples as a deprecated no-op argument

airflow-core/newsfragments/66161.significant.rst

@@ -10,3 +10,9 @@ REST API clients that filtered ``bundle_name`` by ``"dags-folder"`` for
 provider-shipped example DAGs (e.g. ``example_python_operator``) must
 update to the new per-provider bundle names. DAG identifiers are
 unchanged.
+
+The ``include_examples`` argument of ``DagBag`` and ``DagBag.collect_dags``
+is deprecated, has no effect, and will be removed in Airflow 4. A ``DagBag``
+now only ever parses the given folder; example Dags are provided exclusively
+through the per-provider bundles gated by ``[core] load_examples``. Passing
+the argument emits a ``RemovedInAirflow4Warning``.

airflow-core/src/airflow/dag_processing/dagbag.py

@@ -39,6 +39,7 @@
     AirflowClusterPolicyViolation,
     AirflowDagDuplicatedIdException,
     AirflowException,
+    RemovedInAirflow4Warning,
     UnknownExecutorException,
 )
 from airflow.executors.executor_loader import ExecutorLoader
@@ -161,6 +162,16 @@ def _validate_executor_fields(dag: DAG, bundle_name: str | None = None) -> None:
             )
 
 
+def _warn_include_examples_deprecated() -> None:
+    warnings.warn(
+        "The 'include_examples' argument is deprecated, has no effect, and will be removed in "
+        "Airflow 4. Example Dags are loaded through per-provider Dag bundles when the "
+        "[core] load_examples configuration option is enabled.",
+        RemovedInAirflow4Warning,
+        stacklevel=3,
+    )
+
+
 class DagBag(LoggingMixin):
     """
     A dagbag is a collection of dags, parsed out of a folder tree and has high level configuration settings.
@@ -180,6 +191,8 @@ class DagBag(LoggingMixin):
         are not loaded to not run User code in Scheduler.
     :param collect_dags: when True, collects dags during class initialization.
     :param known_pools: If not none, then generate warnings if a Task attempts to use an unknown pool.
+    :param include_examples: Deprecated and ignored; example Dags are loaded through per-provider
+        Dag bundles when the ``[core] load_examples`` configuration option is enabled.
     """
 
     def __init__(
@@ -191,8 +204,11 @@ def __init__(
         known_pools: set[str] | None = None,
         bundle_path: Path | None = None,
         bundle_name: str | None = None,
+        include_examples: bool | ArgNotSet = NOTSET,
     ):
         super().__init__()
+        if is_arg_set(include_examples):
+            _warn_include_examples_deprecated()
         self.bundle_path = bundle_path
         self.bundle_name = bundle_name
 
@@ -448,6 +464,7 @@ def collect_dags(
         dag_folder: str | Path | None = None,
         only_if_updated: bool = True,
         safe_mode: bool = conf.getboolean("core", "DAG_DISCOVERY_SAFE_MODE"),
+        include_examples: bool | ArgNotSet = NOTSET,
     ):
         """
         Look for python modules in a given path, import them, and add them to the dagbag collection.
@@ -461,6 +478,8 @@ def collect_dags(
         un-anchored regexes or gitignore-like glob expressions, depending on
         the ``DAG_IGNORE_FILE_SYNTAX`` configuration parameter.
         """
+        if is_arg_set(include_examples):
+            _warn_include_examples_deprecated()
         self.log.info("Filling up the DagBag from %s", dag_folder)
         dag_folder = dag_folder or self.dag_folder
         # Used to store stats around DagBag processing

airflow-core/tests/unit/dag_processing/test_dagbag.py

@@ -42,7 +42,7 @@
     _capture_with_reraise,
     _validate_executor_fields,
 )
-from airflow.exceptions import UnknownExecutorException
+from airflow.exceptions import RemovedInAirflow4Warning, UnknownExecutorException
 from airflow.executors.executor_loader import ExecutorLoader
 from airflow.models.dag import DagModel
 from airflow.models.dagwarning import DagWarning, DagWarningType
@@ -1397,3 +1397,25 @@ def test_dagbag_no_bundle_path_no_syspath_modification(self, tmp_path):
 
         assert str(tmp_path) not in dag.description
         assert sys.path == syspath_before
+
+
+@pytest.mark.parametrize("include_examples", [True, False])
+def test_dagbag_include_examples_is_deprecated_and_ignored(include_examples, tmp_path):
+    with pytest.warns(RemovedInAirflow4Warning, match="'include_examples' argument is deprecated"):
+        dagbag = DagBag(dag_folder=tmp_path, include_examples=include_examples)
+    assert dagbag.dags == {}
+
+
+@pytest.mark.parametrize("include_examples", [True, False])
+def test_collect_dags_include_examples_is_deprecated_and_ignored(include_examples, tmp_path):
+    dagbag = DagBag(dag_folder=tmp_path, collect_dags=False)
+    with pytest.warns(RemovedInAirflow4Warning, match="'include_examples' argument is deprecated"):
+        dagbag.collect_dags(include_examples=include_examples)
+    assert dagbag.dags == {}
+
+
+def test_dagbag_does_not_warn_when_include_examples_is_not_passed(tmp_path):
+    with warnings.catch_warnings():
+        warnings.simplefilter("error", RemovedInAirflow4Warning)
+        dagbag = DagBag(dag_folder=tmp_path)
+        dagbag.collect_dags()