DagBag: use Path.relative_to for consistent cross-platform behavior (#59785)

Author: Dev-iL

airflow-core/newsfragments/59785.significant.rst

@@ -0,0 +1,10 @@
+Modify the information returned by ``DagBag``
+
+**New behavior:**
+- ``DagBag`` now uses ``Path.relative_to`` for consistent cross-platform behavior.
+- ``FileLoadStat`` now has two additional nullable fields: ``bundle_path`` and ``bundle_name``.
+
+**Backward compatibility:**
+``FileLoadStat`` will no longer produce paths beginning with ``/`` with the meaning of "relative to the dags folder".
+This is a breaking change for any custom code that performs string-based path manipulations relying on this behavior.
+Users are advised to update such code to use ``pathlib.Path``.

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

@@ -97,6 +97,8 @@ class FileLoadStat(NamedTuple):
     :param task_num: Total number of Tasks loaded in this file.
     :param dags: DAGs names loaded in this file.
     :param warning_num: Total number of warnings captured from processing this file.
+    :param bundle_path: The bundle path from DagBag, if any.
+    :param bundle_name: The bundle name from DagBag, if any.
     """
 
     file: str
@@ -105,6 +107,8 @@ class FileLoadStat(NamedTuple):
     task_num: int
     dags: str
     warning_num: int
+    bundle_path: Path | None
+    bundle_name: str | None
 
 
 @contextlib.contextmanager
@@ -647,14 +651,21 @@ def collect_dags(
                 found_dags = self.process_file(filepath, only_if_updated=only_if_updated, safe_mode=safe_mode)
 
                 file_parse_end_dttm = timezone.utcnow()
+                try:
+                    relative_file = Path(filepath).relative_to(Path(self.dag_folder)).as_posix()
+                except ValueError:
+                    # filepath is not under dag_folder (e.g., example DAGs from a different location)
+                    relative_file = Path(filepath).as_posix()
                 stats.append(
                     FileLoadStat(
-                        file=filepath.replace(settings.DAGS_FOLDER, ""),
+                        file=relative_file,
                         duration=file_parse_end_dttm - file_parse_start_dttm,
                         dag_num=len(found_dags),
                         task_num=sum(len(dag.tasks) for dag in found_dags),
                         dags=str([dag.dag_id for dag in found_dags]),
                         warning_num=len(self.captured_warnings.get(filepath, [])),
+                        bundle_path=self.bundle_path,
+                        bundle_name=self.bundle_name,
                     )
                 )
             except Exception as e:

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

@@ -225,8 +225,9 @@ def test_cli_report(self, stdout_capture):
             dag_command.dag_report(args)
             out = temp_stdout.getvalue()
 
-        assert "airflow/example_dags/example_complex.py" in out
-        assert "example_complex" in out
+        data = json.loads(out)
+        assert any(item["file"].endswith("example_complex.py") for item in data)
+        assert any("example_complex" in item["dags"] for item in data)
 
     @conf_vars({("core", "load_examples"): "true"})
     def test_cli_get_dag_details(self, stdout_capture):

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

@@ -368,7 +368,7 @@ def test_safe_mode_heuristic_match(self, tmp_path):
             dagbag = DagBag(include_examples=False, safe_mode=True)
 
         assert len(dagbag.dagbag_stats) == 1
-        assert dagbag.dagbag_stats[0].file == f"/{path.name}"
+        assert dagbag.dagbag_stats[0].file == path.name
 
     def test_safe_mode_heuristic_mismatch(self, tmp_path):
         """
@@ -388,7 +388,65 @@ def test_safe_mode_disabled(self, tmp_path):
         with conf_vars({("core", "dags_folder"): os.fspath(path.parent)}):
             dagbag = DagBag(include_examples=False, safe_mode=False)
         assert len(dagbag.dagbag_stats) == 1
-        assert dagbag.dagbag_stats[0].file == f"/{path.name}"
+        assert dagbag.dagbag_stats[0].file == path.name
+
+    def test_dagbag_stats_file_is_relative_path_with_mixed_separators(self, tmp_path):
+        """
+        Test that dagbag_stats.file contains a relative path even when DAGS_FOLDER
+        and filepath have different path separators (simulates Windows behavior).
+
+        On Windows, settings.DAGS_FOLDER may use forward slashes (e.g., 'C:/foo/dags')
+        while filepath from os.path operations uses backslashes (e.g., 'C:\\foo\\dags\\my_dag.py').
+        This test verifies that path normalization works correctly in such cases.
+
+        See: https://github.com/apache/airflow/issues/XXXXX
+        """
+        path = tmp_path / "testfile.py"
+        path.write_text("# airflow\n# DAG")
+
+        # Simulate the Windows scenario where DAGS_FOLDER has forward slashes
+        # but the filesystem returns paths with backslashes
+        dags_folder_with_forward_slashes = path.parent.as_posix()
+        with conf_vars({("core", "dags_folder"): dags_folder_with_forward_slashes}):
+            dagbag = DagBag(include_examples=False, safe_mode=True)
+
+        assert len(dagbag.dagbag_stats) == 1
+        assert dagbag.dagbag_stats[0].file == path.name
+
+    def test_dagbag_stats_includes_bundle_info(self, tmp_path):
+        """Test that FileLoadStat includes bundle_path and bundle_name from DagBag."""
+        path = tmp_path / "testfile.py"
+        path.write_text("# airflow\n# DAG")
+
+        bundle_path = tmp_path / "bundle"
+        bundle_path.mkdir()
+        bundle_name = "test_bundle"
+
+        with conf_vars({("core", "dags_folder"): os.fspath(path.parent)}):
+            dagbag = DagBag(
+                include_examples=False,
+                safe_mode=True,
+                bundle_path=bundle_path,
+                bundle_name=bundle_name,
+            )
+
+        assert len(dagbag.dagbag_stats) == 1
+        stat = dagbag.dagbag_stats[0]
+        assert stat.bundle_path == bundle_path
+        assert stat.bundle_name == bundle_name
+
+    def test_dagbag_stats_bundle_info_none_when_not_provided(self, tmp_path):
+        """Test that FileLoadStat has None for bundle_path and bundle_name when not provided."""
+        path = tmp_path / "testfile.py"
+        path.write_text("# airflow\n# DAG")
+
+        with conf_vars({("core", "dags_folder"): os.fspath(path.parent)}):
+            dagbag = DagBag(include_examples=False, safe_mode=True)
+
+        assert len(dagbag.dagbag_stats) == 1
+        stat = dagbag.dagbag_stats[0]
+        assert stat.bundle_path is None
+        assert stat.bundle_name is None
 
     def test_process_file_that_contains_multi_bytes_char(self, tmp_path):
         """