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))

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

@@ -42,6 +42,7 @@
 from airflow.serialization.serialized_objects import SerializedDAG
 from airflow.utils import timezone as tz
 from airflow.utils.session import create_session
+from scripts.ci.pre_commit.common_precommit_utils import AIRFLOW_ROOT_PATH
 
 from tests_common.test_utils import db
 from tests_common.test_utils.asserts import assert_queries_count
@@ -52,6 +53,12 @@
 pytestmark = pytest.mark.db_test
 
 example_dags_folder = pathlib.Path(airflow.example_dags.__path__[0])  # type: ignore[attr-defined]
+try:
+    import system.standard
+
+    example_standard_dags_folder = pathlib.Path(system.standard.__path__[0])  # type: ignore[attr-defined]
+except ImportError:
+    example_standard_dags_folder = pathlib.Path(airflow.example_dags.__path__[0])  # type: ignore[attr-defined]
 
 PY311 = sys.version_info >= (3, 11)
 
@@ -359,13 +366,16 @@ def process_file(self, filepath, only_if_updated=True, safe_mode=True):
         ("file_to_load", "expected"),
         (
             pytest.param(
-                pathlib.Path(example_dags_folder) / "example_bash_operator.py",
-                {"example_bash_operator": "airflow/example_dags/example_bash_operator.py"},
+                pathlib.Path(example_standard_dags_folder) / "example_bash_operator.py",
+                {
+                    "example_bash_operator": f"{example_standard_dags_folder.relative_to(AIRFLOW_ROOT_PATH) / 'example_bash_operator.py'}"
+                },
                 id="example_bash_operator",
             ),
         ),
     )
     def test_get_dag_registration(self, file_to_load, expected):
+        pytest.importorskip("system.standard")
         dagbag = DagBag(dag_folder=os.devnull, include_examples=False)
         dagbag.process_file(os.fspath(file_to_load))
         for dag_id, path in expected.items():
@@ -420,7 +430,7 @@ def test_refresh_py_dag(self, mock_dagmodel, tmp_path):
         Test that we can refresh an ordinary .py DAG
         """
         dag_id = "example_bash_operator"
-        fileloc = str(example_dags_folder / "example_bash_operator.py")
+        fileloc = str(example_standard_dags_folder / "example_bash_operator.py")
 
         mock_dagmodel.return_value = DagModel()
         mock_dagmodel.return_value.last_expired = datetime.max.replace(tzinfo=timezone.utc)

clients/python/test_python_client.py

@@ -69,7 +69,7 @@
 
 # Make sure in the [core] section, the  `load_examples` config is set to True in your airflow.cfg
 # or AIRFLOW__CORE__LOAD_EXAMPLES environment variable set to True
-DAG_ID = "example_bash_operator"
+DAG_ID = "example_simplest_dag"
 
 
 # Enter a context with an instance of the API client

devel-common/src/docs/provider_conf.py

@@ -156,15 +156,8 @@
     "operators/_partials",
     "_api/airflow/index.rst",
     "_api/airflow/providers/index.rst",
-    "_api/airflow/providers/apache/index.rst",
-    "_api/airflow/providers/atlassian/index.rst",
-    "_api/airflow/providers/cncf/index.rst",
-    "_api/airflow/providers/common/index.rst",
-    "_api/airflow/providers/common/messaging/providers/base_provider/index.rst",
-    "_api/airflow/providers/common/messaging/providers/sqs/index.rst",
-    "_api/airflow/providers/dbt/index.rst",
-    "_api/airflow/providers/microsoft/index.rst",
     "_api/docs/conf",
+    *[f"_api/airflow/providers/{subpackage}/index.rst" for subpackage in empty_subpackages],
     *[f"_api/system/{subpackage}/index.rst" for subpackage in empty_subpackages],
     *[f"_api/tests/system/{subpackage}/index.rst" for subpackage in empty_subpackages],
 ]

devel-common/src/sphinx_exts/docs_build/docs_builder.py

@@ -120,6 +120,13 @@ def _src_dir(self) -> Path:
         console.print(f"[red]Unknown package name: {self.package_name}")
         sys.exit(1)
 
+    @property
+    def pythonpath(self) -> list[Path]:
+        path = []
+        if (self._src_dir.parent / "tests").exists():
+            path.append(self._src_dir.parent.joinpath("tests").resolve())
+        return path
+
     @property
     def _generated_api_dir(self) -> Path:
         return self._build_dir.resolve() / "_api"
@@ -167,6 +174,8 @@ def check_spelling(self, verbose: bool) -> tuple[list[SpellingError], list[DocBu
             console.print("[yellow]Command to run:[/] ", " ".join([shlex.quote(arg) for arg in build_cmd]))
         env = os.environ.copy()
         env["AIRFLOW_PACKAGE_NAME"] = self.package_name
+        if self.pythonpath:
+            env["PYTHONPATH"] = ":".join([path.as_posix() for path in self.pythonpath])
         if verbose:
             console.print(
                 f"[bright_blue]{self.package_name:60}:[/] The output is hidden until an error occurs."
@@ -246,6 +255,8 @@ def build_sphinx_docs(self, verbose: bool) -> list[DocBuildError]:
             console.print("[yellow]Command to run:[/] ", " ".join([shlex.quote(arg) for arg in build_cmd]))
         env = os.environ.copy()
         env["AIRFLOW_PACKAGE_NAME"] = self.package_name
+        if self.pythonpath:
+            env["PYTHONPATH"] = ":".join([path.as_posix() for path in self.pythonpath])
         if verbose:
             console.print(
                 f"[bright_blue]{self.package_name:60}:[/] Running sphinx. "