Add back missing [sources] link in generated documentation's includes (apache#49978)

* Add back missing `[sources]` link in generated documentation's includes

The exampleinclude of ours used to have generated "sources" link to link
to sources of the examples. This was however gone after recent doc
build refactoring.

The problem was the example directive include import and lack of
tests folder available on the python path. Both issues fixed in this
PR.

Fixes: apache#49893

* fixup! Add back missing `[sources]` link in generated documentation's includes

Author: potiuk

Dockerfile.ci

@@ -949,6 +949,12 @@ function environment_initialization() {
 
     cd "${AIRFLOW_SOURCES}"
 
+    # Temporarily add /opt/airflow/providers/standard/tests to PYTHONPATH in order to see example dags
+    # in the UI when testing in Breeze. This might be solved differently in the future
+    if [[ -d /opt/airflow/providers/standard/tests  ]]; then
+        export PYTHONPATH=${PYTHONPATH=}:/opt/airflow/providers/standard/tests
+    fi
+
     if [[ ${START_AIRFLOW:="false"} == "true" || ${START_AIRFLOW} == "True" ]]; then
         export AIRFLOW__CORE__LOAD_EXAMPLES=${LOAD_EXAMPLES}
         wait_for_asset_compilation

airflow-core/docs/tutorial/taskflow.rst

@@ -269,7 +269,7 @@ system-level packages. TaskFlow supports multiple execution environments to isol
 Creates a temporary virtualenv at task runtime. Great for experimental or dynamic tasks, but may have cold start
 overhead.
 
-.. exampleinclude:: /../src/airflow/example_dags/example_python_decorator.py
+.. exampleinclude:: /../../providers/standard/tests/system/standard/example_python_decorator.py
     :language: python
     :dedent: 4
     :start-after: [START howto_operator_python_venv]
@@ -283,7 +283,7 @@ overhead.
 
 Executes the task using a pre-installed Python interpreter — ideal for consistent environments or shared virtualenvs.
 
-.. exampleinclude:: /../src/airflow/example_dags/example_python_decorator.py
+.. exampleinclude:: /../../providers/standard/tests/system/standard/example_python_decorator.py
     :language: python
     :dedent: 4
     :start-after: [START howto_operator_external_python]
@@ -333,7 +333,7 @@ Using Sensors
 Use ``@task.sensor`` to build lightweight, reusable sensors using Python functions. These support both poke and reschedule
 modes.
 
-.. exampleinclude:: /../src/airflow/example_dags/example_sensor_decorator.py
+.. exampleinclude:: /../../providers/standard/tests/system/standard/example_sensor_decorator.py
     :language: python
     :start-after: [START tutorial]
     :end-before: [END tutorial]

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

@@ -16,6 +16,7 @@
 # under the License.
 from __future__ import annotations
 
+import os
 from typing import TYPE_CHECKING
 
 from airflow.configuration import conf
@@ -34,6 +35,7 @@
     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):
@@ -80,6 +82,25 @@ 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."""
 
@@ -112,6 +133,11 @@ 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/models/dagbag.py

@@ -586,6 +586,14 @@ 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/src/airflow/utils/cli.py

@@ -270,7 +270,6 @@ def get_dag(bundle_names: list | None, dag_id: str, from_db: bool = False) -> DA
 
     bundle_names = bundle_names or []
     dag: DAG | None = None
-
     if from_db:
         dagbag = DagBag(read_dags_from_db=True)
         dag = dagbag.get_dag(dag_id)  # get_dag loads from the DB as requested

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

@@ -28,8 +28,8 @@
 
 pytestmark = pytest.mark.db_test
 
-EXAMPLE_DAG_FILE = AIRFLOW_CORE_SOURCES_PATH / "airflow" / "example_dags" / "example_bash_operator.py"
-TEST_DAG_ID = "example_bash_operator"
+EXAMPLE_DAG_FILE = AIRFLOW_CORE_SOURCES_PATH / "airflow" / "example_dags" / "example_simplest_dag.py"
+TEST_DAG_ID = "example_simplest_dag"
 NOT_READABLE_DAG_ID = "latest_only_with_trigger"
 TEST_MULTIPLE_DAGS_ID = "asset_produces_1"
 

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

@@ -23,6 +23,7 @@
 
 from airflow.utils.file import list_py_file_paths
 
+import system.standard
 from tests_common.test_utils.config import conf_vars
 from tests_common.test_utils.db import clear_db_dags, parse_and_sync_to_db
 
@@ -37,8 +38,10 @@
 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
+    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(next(iter(system.standard.__path__)))) if include_examples else 0)
     )
 
 

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

@@ -39,9 +39,9 @@
 
 # Example bash operator located here: airflow/example_dags/example_bash_operator.py
 EXAMPLE_DAG_FILE = (
-    AIRFLOW_REPO_ROOT_PATH / "airflow-core" / "src" / "airflow" / "example_dags" / "example_bash_operator.py"
+    AIRFLOW_REPO_ROOT_PATH / "airflow-core" / "src" / "airflow" / "example_dags" / "example_simplest_dag.py"
 )
-TEST_DAG_ID = "example_bash_operator"
+TEST_DAG_ID = "example_simplest_dag"
 
 
 @pytest.fixture

airflow-core/tests/unit/cli/commands/test_task_command.py

@@ -47,6 +47,7 @@
 from airflow.utils.state import State, TaskInstanceState
 from airflow.utils.types import DagRunTriggeredByType, DagRunType
 
+from tests_common.test_utils.config import conf_vars
 from tests_common.test_utils.db import clear_db_runs, parse_and_sync_to_db
 
 pytestmark = pytest.mark.db_test
@@ -74,7 +75,6 @@ def move_back(old_path, new_path):
     shutil.move(new_path, old_path)
 
 
-# TODO: Check if tests needs side effects - locally there's missing DAG
 class TestCliTasks:
     run_id = "TEST_RUN_ID"
     dag_id = "example_python_operator"
@@ -91,7 +91,7 @@ def setup_class(cls):
         cls.parser = cli_parser.get_parser()
         clear_db_runs()
 
-        cls.dagbag = DagBag(read_dags_from_db=True)
+        cls.dagbag = DagBag(read_dags_from_db=True, include_examples=True)
         cls.dag = cls.dagbag.get_dag(cls.dag_id)
         data_interval = cls.dag.timetable.infer_manual_data_interval(run_after=DEFAULT_DATE)
         cls.dag_run = cls.dag.create_dagrun(
@@ -108,6 +108,7 @@ def setup_class(cls):
     def teardown_class(cls) -> None:
         clear_db_runs()
 
+    @conf_vars({("core", "load_examples"): "true"})
     @pytest.mark.execution_timeout(120)
     def test_cli_list_tasks(self):
         for dag_id in self.dagbag.dags:

airflow-core/tests/unit/jobs/test_scheduler_job.py

@@ -37,7 +37,6 @@
 from sqlalchemy import func, select, update
 from sqlalchemy.orm import joinedload
 
-import airflow.example_dags
 from airflow import settings
 from airflow.assets.manager import AssetManager
 from airflow.callbacks.callback_requests import DagCallbackRequest, TaskCallbackRequest
@@ -75,6 +74,7 @@
 from airflow.utils.thread_safe_dict import ThreadSafeDict
 from airflow.utils.types import DagRunTriggeredByType, DagRunType
 
+from system import standard
 from tests_common.test_utils.asserts import assert_queries_count
 from tests_common.test_utils.config import conf_vars, env_vars
 from tests_common.test_utils.db import (
@@ -104,7 +104,7 @@
 ELASTIC_DAG_FILE = os.path.join(PERF_DAGS_FOLDER, "elastic_dag.py")
 
 TEST_DAG_FOLDER = os.environ["AIRFLOW__CORE__DAGS_FOLDER"]
-EXAMPLE_DAGS_FOLDER = airflow.example_dags.__path__[0]
+EXAMPLE_STANDARD_DAGS_FOLDER = standard.__path__[0]
 DEFAULT_DATE = timezone.datetime(2016, 1, 1)
 DEFAULT_LOGICAL_DATE = timezone.coerce_datetime(DEFAULT_DATE)
 TRY_NUMBER = 1
@@ -5707,7 +5707,7 @@ def test_find_and_purge_task_instances_without_heartbeats_nothing(self):
 
     @pytest.mark.usefixtures("testing_dag_bundle")
     def test_find_and_purge_task_instances_without_heartbeats(self, session, create_dagrun):
-        dagfile = os.path.join(EXAMPLE_DAGS_FOLDER, "example_branch_operator.py")
+        dagfile = os.path.join(EXAMPLE_STANDARD_DAGS_FOLDER, "example_branch_operator.py")
         dagbag = DagBag(dagfile)
         dag = dagbag.get_dag("example_branch_operator")
         dm = LazyDeserializedDAG(data=SerializedDAG.to_dict(dag))
@@ -5773,7 +5773,7 @@ def test_task_instance_heartbeat_timeout_message(self, session, create_dagrun):
         """
         Check that the task instance heartbeat timeout message comes out as expected
         """
-        dagfile = os.path.join(EXAMPLE_DAGS_FOLDER, "example_branch_operator.py")
+        dagfile = os.path.join(EXAMPLE_STANDARD_DAGS_FOLDER, "example_branch_operator.py")
         dagbag = DagBag(dagfile)
         dag = dagbag.get_dag("example_branch_operator")
         dm = LazyDeserializedDAG(data=SerializedDAG.to_dict(dag))