Restore include_examples parameter to DagBag for backwards compatibility

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

@@ -172,6 +172,16 @@ class DagBag(LoggingMixin):
     that one system can run multiple, independent settings sets.
 
     :param dag_folder: the folder to scan to find DAGs
+    :param include_examples: whether to include the examples that ship
+        with airflow or not.
+
+        .. deprecated::
+            Passing this explicitly is deprecated. Example DAGs are now loaded via the
+            DAG bundle system (see ``[core] LOAD_EXAMPLES``) rather than by this class
+            scanning ``airflow.example_dags`` directly. This parameter is kept only for
+            backwards compatibility with callers that construct ``DagBag`` directly
+            (e.g. custom scripts, DAG validation tests) and will be removed in a future
+            release.
     :param safe_mode: when ``False``, scans all python modules for dags.
         When ``True`` uses heuristics (files containing ``DAG`` and ``airflow`` strings)
         to filter python modules to scan for dags.
@@ -185,6 +195,7 @@ class DagBag(LoggingMixin):
     def __init__(
         self,
         dag_folder: str | Path | None = None,  # todo AIP-66: rename this to path
+        include_examples: bool | ArgNotSet = NOTSET,
         safe_mode: bool | ArgNotSet = NOTSET,
         load_op_links: bool = True,
         collect_dags: bool = True,
@@ -196,6 +207,18 @@ def __init__(
         self.bundle_path = bundle_path
         self.bundle_name = bundle_name
 
+        if is_arg_set(include_examples):
+            warnings.warn(
+                "Passing `include_examples` to DagBag is deprecated and will be removed in a "
+                "future release. Example DAGs are now loaded via the DAG bundle system; see the "
+                "`[core] LOAD_EXAMPLES` configuration option.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+            include_examples = bool(include_examples)
+        else:
+            include_examples = conf.getboolean("core", "LOAD_EXAMPLES")
+
         dag_folder = dag_folder or settings.DAGS_FOLDER
         self.dag_folder = dag_folder
         self.dags: dict[str, DAG] = {}
@@ -215,6 +238,7 @@ def __init__(
         if collect_dags:
             self.collect_dags(
                 dag_folder=dag_folder,
+                include_examples=include_examples,
                 safe_mode=(
                     safe_mode if is_arg_set(safe_mode) else conf.getboolean("core", "DAG_DISCOVERY_SAFE_MODE")
                 ),
@@ -447,6 +471,7 @@ def collect_dags(
         self,
         dag_folder: str | Path | None = None,
         only_if_updated: bool = True,
+        include_examples: bool = conf.getboolean("core", "LOAD_EXAMPLES"),
         safe_mode: bool = conf.getboolean("core", "DAG_DISCOVERY_SAFE_MODE"),
     ):
         """
@@ -472,6 +497,12 @@ def collect_dags(
         registry = get_importer_registry()
         files_to_parse = registry.list_dag_files(dag_folder, safe_mode=safe_mode)
 
+        if include_examples:
+            from airflow import example_dags
+
+            example_dag_folder = next(iter(example_dags.__path__))
+            files_to_parse.extend(registry.list_dag_files(example_dag_folder, safe_mode=safe_mode))
+
         for filepath in files_to_parse:
             try:
                 file_parse_start_dttm = timezone.utcnow()
@@ -542,7 +573,18 @@ def __init__(self, *args, bundle_path: Path | None = None, **kwargs):
         if str(bundle_path) not in sys.path:
             sys.path.append(str(bundle_path))
 
+        # Warn if user explicitly set include_examples=True, since bundles never contain examples
+        # (example DAGs are loaded via their own dedicated bundle, see DagBundlesManager).
+        if kwargs.get("include_examples") is True:
+            warnings.warn(
+                "include_examples=True is ignored for BundleDagBag. "
+                "Bundles do not contain example DAGs, so include_examples is always False.",
+                UserWarning,
+                stacklevel=2,
+            )
+
         kwargs["bundle_path"] = bundle_path
+        kwargs["include_examples"] = False
         super().__init__(*args, **kwargs)
 
 

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

@@ -1397,3 +1397,43 @@ 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
+
+
+class TestDagBagIncludeExamples:
+    """Regression tests for https://github.com/apache/airflow/issues/70283."""
+
+    def test_include_examples_not_passed_uses_config_no_warning(self):
+        """Omitting include_examples entirely must not warn and must fall back to config."""
+        with conf_vars({("core", "LOAD_EXAMPLES"): "False"}):
+            with warnings.catch_warnings(record=True) as caught:
+                warnings.simplefilter("always")
+                dagbag = DagBag(dag_folder=os.devnull, collect_dags=True)
+            assert not any(issubclass(w.category, DeprecationWarning) for w in caught)
+            assert dagbag.dags == {}
+
+    def test_include_examples_explicit_true_loads_examples_and_warns(self):
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            dagbag = DagBag(dag_folder=os.devnull, include_examples=True, collect_dags=True)
+        assert any(issubclass(w.category, DeprecationWarning) for w in caught)
+        assert len(dagbag.dags) > 0  # example DAGs were loaded
+
+    def test_include_examples_explicit_false_warns_but_loads_nothing(self):
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            dagbag = DagBag(dag_folder=os.devnull, include_examples=False, collect_dags=True)
+        assert any(issubclass(w.category, DeprecationWarning) for w in caught)
+        assert dagbag.dags == {}
+
+    def test_bundle_dagbag_ignores_include_examples_true_with_warning(self, tmp_path):
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            dagbag = BundleDagBag(
+                dag_folder=os.devnull,
+                bundle_path=tmp_path,
+                include_examples=True,
+                collect_dags=True,
+            )
+        assert any(issubclass(w.category, UserWarning) for w in caught)
+        assert dagbag.dags == {}
+        sys.path.remove(str(tmp_path))