Bring back "standard" example dags to the airflow-core package (#51192)

As we are working on a longer-term solution for example-dags, the
short-term one is to bring the standard example_dags back to the
airflow-core via symbolic link to the standard package folder.

This makes the dags to be copied to the airflow-core from the latest
main version (but those dags don't change) while not requiring to
duplicate the dags and keep them in standard provider where the
documentation for them is kept.

Co-authored-by: Kaxil Naik <[email protected]>

Author: potiuk

airflow-core/src/airflow/dag_processing/bundles/manager.py

@@ -16,7 +16,6 @@
 # under the License.
 from __future__ import annotations
 
-import os
 from typing import TYPE_CHECKING
 
 from airflow.configuration import conf
@@ -35,7 +34,6 @@
     from airflow.dag_processing.bundles.base import BaseDagBundle
 
 _example_dag_bundle_name = "example_dags"
-_example_standard_dag_bundle_name = "example_standard_dags"
 
 
 def _bundle_item_exc(msg):
@@ -82,25 +80,6 @@ def _add_example_dag_bundle(config_list):
     )
 
 
-def _add_example_standard_dag_bundle(config_list):
-    # TODO(potiuk): make it more generic - for now we only add standard example_dags if they are locally available
-    try:
-        from system import standard
-    except ImportError:
-        return
-
-    example_dag_folder = next(iter(standard.__path__))
-    config_list.append(
-        {
-            "name": _example_standard_dag_bundle_name,
-            "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
-            "kwargs": {
-                "path": example_dag_folder,
-            },
-        }
-    )
-
-
 class DagBundlesManager(LoggingMixin):
     """Manager for DAG bundles."""
 
@@ -133,11 +112,6 @@ def parse_config(self) -> None:
         _validate_bundle_config(config_list)
         if conf.getboolean("core", "LOAD_EXAMPLES"):
             _add_example_dag_bundle(config_list)
-            if (
-                os.environ.get("BREEZE", "").lower() == "true"
-                or os.environ.get("_IN_UNIT_TESTS", "").lower() == "true"
-            ):
-                _add_example_standard_dag_bundle(config_list)
 
         for cfg in config_list:
             name = cfg["name"]

airflow-core/src/airflow/example_dags/standard

@@ -0,0 +1 @@
+../../../../providers/standard/tests/system/standard/

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

@@ -598,14 +598,6 @@ def collect_dags(
             example_dag_folder = next(iter(example_dags.__path__))
 
             files_to_parse.extend(list_py_file_paths(example_dag_folder, safe_mode=safe_mode))
-            try:
-                from system import standard
-
-                example_dag_folder_standard = next(iter(standard.__path__))
-                files_to_parse.extend(list_py_file_paths(example_dag_folder_standard, safe_mode=safe_mode))
-            except ImportError:
-                # Nothing happens - this should only work during tests
-                pass
 
         for filepath in files_to_parse:
             try:

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_report.py

@@ -23,7 +23,6 @@
 
 from airflow.utils.file import list_py_file_paths
 
-from tests_common.pytest_plugin import AIRFLOW_ROOT_PATH
 from tests_common.test_utils.config import conf_vars
 from tests_common.test_utils.db import clear_db_dags, parse_and_sync_to_db
 
@@ -34,18 +33,12 @@
 TEST_DAG_FOLDER_INVALID = "/invalid/path"
 TEST_DAG_FOLDER_INVALID_2 = "/root/airflow/tests/dags/"
 
-STANDARD_PROVIDER_SYSTEM_TESTS_PATH = (
-    AIRFLOW_ROOT_PATH / "providers" / "standard" / "tests" / "system" / "standard"
-)
-
 
 def get_corresponding_dag_file_count(dir: str, include_examples: bool = True) -> int:
     from airflow import example_dags
 
-    return (
-        len(list_py_file_paths(directory=dir))
-        + (len(list_py_file_paths(next(iter(example_dags.__path__)))) if include_examples else 0)
-        + (len(list_py_file_paths(STANDARD_PROVIDER_SYSTEM_TESTS_PATH.as_posix())) if include_examples else 0)
+    return len(list_py_file_paths(directory=dir)) + (
+        len(list_py_file_paths(next(iter(example_dags.__path__)))) if include_examples else 0
     )
 
 

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

@@ -52,7 +52,7 @@
 
 pytestmark = pytest.mark.db_test
 
-example_dags_folder = AIRFLOW_ROOT_PATH / "providers" / "standard" / "tests" / "system" / "standard"
+example_dags_folder = AIRFLOW_ROOT_PATH / "airflow-core" / "src" / "airflow" / "example_dags" / "standard"
 
 PY311 = sys.version_info >= (3, 11)