Fix DAG import errors silently dropped when the bundle is re-instantiated inside the parsing transaction

airflow-core/newsfragments/69845.bugfix.rst

@@ -0,0 +1 @@
+Fix DAG import errors being silently dropped (never shown in the *Import Errors* view). While recording an import error, ``_update_import_errors`` called ``ParseImportError.full_file_path()`` to build the import-error listener argument. ``full_file_path()`` re-instantiates the DAG bundle and, for some bundle types (e.g. the Git bundle), resolves an Airflow Connection; doing so inside the DAG-parsing transaction could roll back the not-yet-flushed import error row, leaving the DAG marked as broken with nothing shown in the UI. The already-known bundle path is now threaded down to ``_update_import_errors`` (from both the DAG processor and ``sync_bag_to_db``) and used to build the listener filename directly, so the bundle is no longer re-instantiated inside the transaction. Additionally, when no listener plugin is registered, the listener arguments are no longer computed at all.

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

@@ -71,6 +71,7 @@
 
 if TYPE_CHECKING:
     from collections.abc import Collection, Iterable, Iterator
+    from pathlib import Path
 
     from sqlalchemy.orm import Session
     from sqlalchemy.sql import Select
@@ -397,9 +398,36 @@ def _update_import_errors(
     bundle_name: str,
     import_errors: dict[tuple[str, str], str],
     session: Session,
+    *,
+    bundle_path: Path | None = None,
 ):
     from airflow.listeners.listener import get_listener_manager
 
+    def _listener_filename(
+        relative_fileloc: str, bundle_name_: str, import_error: ParseImportError | None = None
+    ) -> str | None:
+        """
+        Build the file path handed to import-error listeners.
+
+        When the bundle path is known (the DAG processor threads it down from the
+        already-initialised bundle) we build the path directly. This avoids calling
+        ``ParseImportError.full_file_path()``, which re-instantiates the DAG bundle --
+        and, for some bundle types (e.g. the Git bundle), resolves a Connection --
+        inside this parsing transaction, where it can roll back the pending import
+        error row. Falls back to ``full_file_path()`` only when the bundle path was
+        not provided.
+        """
+        if bundle_path is not None:
+            return str(bundle_path / relative_fileloc)
+        if import_error is None:
+            import_error = session.scalar(
+                select(ParseImportError).where(
+                    ParseImportError.bundle_name == bundle_name_,
+                    ParseImportError.filename == relative_fileloc,
+                )
+            )
+        return import_error.full_file_path() if import_error is not None else None
+
     # Check existing import errors BEFORE deleting, so we can determine if we should update or create
     existing_import_error_files = set(
         session.execute(select(ParseImportError.bundle_name, ParseImportError.filename))
@@ -436,16 +464,19 @@ def _update_import_errors(
             # sending notification when an existing dag import error occurs
             try:
                 # todo: make listener accept bundle_name and relative_filename
-                import_error = session.scalar(
-                    select(ParseImportError).where(
-                        ParseImportError.bundle_name == bundle_name_,
-                        ParseImportError.filename == relative_fileloc,
-                    )
-                )
-                if import_error is not None:
-                    get_listener_manager().hook.on_existing_dag_import_error(
-                        filename=import_error.full_file_path(), stacktrace=stacktrace
-                    )
+                # When no listener plugin is registered, skip building the listener
+                # arguments entirely: for callers that do not thread ``bundle_path``
+                # down, the fallback would otherwise query ``ParseImportError`` and
+                # call the bundle-instantiating ``full_file_path()`` inside this
+                # transaction just to feed a hook nobody listens to. The check stays
+                # inside the try block so a listener-manager failure is logged rather
+                # than aborting the recording of import errors.
+                if get_listener_manager().has_listeners:
+                    filename = _listener_filename(relative_fileloc, bundle_name_)
+                    if filename is not None:
+                        get_listener_manager().hook.on_existing_dag_import_error(
+                            filename=filename, stacktrace=stacktrace
+                        )
             except Exception:
                 log.exception("error calling listener")
         else:
@@ -458,9 +489,12 @@ def _update_import_errors(
             session.add(import_error)
             # sending notification when a new dag import error occurs
             try:
-                get_listener_manager().hook.on_new_dag_import_error(
-                    filename=import_error.full_file_path(), stacktrace=stacktrace
-                )
+                if get_listener_manager().has_listeners:
+                    filename = _listener_filename(relative_fileloc, bundle_name_, import_error)
+                    if filename is not None:
+                        get_listener_manager().hook.on_new_dag_import_error(
+                            filename=filename, stacktrace=stacktrace
+                        )
             except Exception:
                 log.exception("error calling listener")
         session.execute(
@@ -494,6 +528,7 @@ def update_dag_parsing_results_in_db(
         DagWarningType.RUNTIME_VARYING_VALUE,
     ),
     files_parsed: set[tuple[str, str]] | None = None,
+    bundle_path: Path | None = None,
 ):
     """
     Update everything to do with DAG parsing in the DB.
@@ -515,6 +550,10 @@ def update_dag_parsing_results_in_db(
     :param files_parsed: Set of (bundle_name, relative_fileloc) tuples for all files that were parsed.
         If None, will be inferred from dags and import_errors. Passing this explicitly ensures that
         import errors are cleared for files that were parsed but no longer contain DAGs.
+    :param bundle_path: Filesystem path of the already-initialised DAG bundle. When provided, it is
+        used to build the file path passed to import-error listeners without re-instantiating the
+        bundle inside this transaction. If None, ``ParseImportError.full_file_path()`` is used as a
+        fallback.
     """
     # Retry 'DAG.bulk_write_to_db' & 'SerializedDagModel.bulk_sync_to_db' in case
     # of any Operational Errors
@@ -571,6 +610,7 @@ def update_dag_parsing_results_in_db(
             bundle_name=bundle_name,
             import_errors=import_errors,
             session=session,
+            bundle_path=bundle_path,
         )
     except Exception:
         log.exception("Error logging import errors!")

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

@@ -579,4 +579,5 @@ def sync_bag_to_db(
         session=session,
         version_data=version_data,
         files_parsed=files_parsed,
+        bundle_path=dagbag.bundle_path,
     )

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

@@ -1243,6 +1243,7 @@ def handle_parsing_result(
                     parsing_result=proc.parsing_result,
                     run_duration=run_duration,
                     relative_fileloc=str(file.rel_path),
+                    bundle_path=file.bundle_path,
                     session=session,
                 )
             except Exception:
@@ -1276,6 +1277,7 @@ def persist_parsing_result(
         run_duration: float,
         relative_fileloc: str | None,
         session: Session,
+        bundle_path: Path | None = None,
     ) -> None:
         """Persist parsed DAG data to the metadata database."""
         import_errors: dict[tuple[str, str], str] = {}
@@ -1305,6 +1307,7 @@ def persist_parsing_result(
             warnings=set(warnings),
             session=session,
             files_parsed=files_parsed,
+            bundle_path=bundle_path,
         )
 
     def _collect_results(self):

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

@@ -22,6 +22,7 @@
 import warnings
 from collections.abc import Generator
 from datetime import timedelta
+from pathlib import Path
 from typing import TYPE_CHECKING
 from unittest import mock
 from unittest.mock import patch
@@ -39,6 +40,7 @@
     _get_latest_runs_stmt,
     _get_latest_runs_stmt_partitioned,
     _update_dag_tags,
+    _update_import_errors,
     update_dag_parsing_results_in_db,
 )
 from airflow.exceptions import SerializationError
@@ -905,6 +907,156 @@ def test_serialized_dag_errors_are_import_errors(
         assert len(dag_import_error_listener.existing) == 0
         assert dag_import_error_listener.new["abc.py"] == import_error.stacktrace
 
+    @patch.object(ParseImportError, "full_file_path")
+    @pytest.mark.usefixtures("clean_db")
+    def test_import_error_listener_uses_bundle_path_without_full_file_path(
+        self, mock_full_path, session, dag_import_error_listener, testing_dag_bundle
+    ):
+        """
+        Import errors are persisted and the listener filename is built from ``bundle_path``.
+
+        ``ParseImportError.full_file_path()`` re-instantiates the DAG bundle (and, for some
+        bundle types such as the Git bundle, resolves an Airflow Connection); calling it inside
+        the parsing transaction could roll back the not-yet-flushed import error row, leaving the
+        DAG marked as broken while nothing showed up in the UI. When the caller threads the
+        already-known ``bundle_path`` down, the filename is built directly and
+        ``full_file_path()`` must not be called at all.
+        """
+        bundle_path = Path("/opt/airflow/bundles/testing")
+        import_errors = {("testing", "abc.py"): "AnImportError"}
+        _update_import_errors(
+            files_parsed={("testing", "abc.py")},
+            bundle_name="testing",
+            import_errors=import_errors,
+            session=session,
+            bundle_path=bundle_path,
+        )
+        session.flush()
+
+        # The bundle-instantiating full_file_path() must not be used when bundle_path is given.
+        mock_full_path.assert_not_called()
+
+        # The import error is still persisted.
+        persisted = session.scalars(select(ParseImportError)).all()
+        assert len(persisted) == 1
+        assert persisted[0].bundle_name == "testing"
+        assert persisted[0].filename == "abc.py"
+        assert persisted[0].stacktrace == "AnImportError"
+
+        # The listener still received the correct filename, built from bundle_path.
+        assert dag_import_error_listener.new[str(bundle_path / "abc.py")] == "AnImportError"
+
+    @patch.object(ParseImportError, "full_file_path")
+    @pytest.mark.usefixtures("clean_db")
+    def test_existing_import_error_listener_uses_bundle_path_without_full_file_path(
+        self, mock_full_path, session, dag_import_error_listener, testing_dag_bundle
+    ):
+        """
+        The existing-error branch also builds the listener filename from ``bundle_path``.
+
+        When an import error row already exists for the file, ``on_existing_dag_import_error``
+        is notified instead of ``on_new_dag_import_error``; that branch previously ran an extra
+        ``select(ParseImportError)`` and called ``full_file_path()`` just to build the listener
+        argument. With ``bundle_path`` threaded down, neither must happen.
+        """
+        mock_full_path.side_effect = AssertionError(
+            "full_file_path() must not be called when bundle_path is provided"
+        )
+        bundle_path = Path("/opt/airflow/bundles/testing")
+        session.add(
+            ParseImportError(
+                filename="abc.py", bundle_name="testing", timestamp=tz.utcnow(), stacktrace="OldError"
+            )
+        )
+        session.flush()
+
+        _update_import_errors(
+            files_parsed={("testing", "abc.py")},
+            bundle_name="testing",
+            import_errors={("testing", "abc.py"): "NewImportError"},
+            session=session,
+            bundle_path=bundle_path,
+        )
+        session.flush()
+
+        # The existing row is updated in place.
+        persisted = session.scalars(select(ParseImportError)).all()
+        assert len(persisted) == 1
+        assert persisted[0].stacktrace == "NewImportError"
+
+        mock_full_path.assert_not_called()
+
+        # The existing-error listener received the bundle_path-based filename.
+        assert dag_import_error_listener.new == {}
+        assert dag_import_error_listener.existing[str(bundle_path / "abc.py")] == "NewImportError"
+
+    @patch.object(ParseImportError, "full_file_path")
+    @pytest.mark.usefixtures("clean_db")
+    def test_sync_bag_to_db_does_not_call_full_file_path(
+        self, mock_full_path, tmp_path, session, dag_import_error_listener, testing_dag_bundle
+    ):
+        """
+        ``sync_bag_to_db`` threads ``dagbag.bundle_path`` down to ``_update_import_errors``.
+
+        This exercises a production caller end-to-end (instead of the private helper), so it
+        fails if any caller in the chain stops forwarding ``bundle_path`` and the code falls
+        back to the bundle-instantiating ``full_file_path()`` inside the parsing transaction.
+        """
+        from airflow.dag_processing.dagbag import DagBag, sync_bag_to_db
+
+        mock_full_path.side_effect = AssertionError(
+            "full_file_path() must not be called when bundle_path is provided"
+        )
+        bundle_path = tmp_path / "bundle"
+        bundle_path.mkdir()
+        dag_file = bundle_path / "broken.py"
+        # Must mention both "airflow" and "dag" or DAG_DISCOVERY_SAFE_MODE skips the file.
+        dag_file.write_text("from airflow.sdk import DAG\nraise ImportError('broken dag')")
+
+        dagbag = DagBag(dag_folder=str(dag_file), bundle_path=bundle_path, bundle_name="testing")
+        assert dagbag.import_errors
+
+        sync_bag_to_db(dagbag, "testing", None, session=session)
+        session.flush()
+
+        persisted = session.scalars(select(ParseImportError)).all()
+        assert len(persisted) == 1
+        assert persisted[0].bundle_name == "testing"
+        assert persisted[0].filename == "broken.py"
+
+        mock_full_path.assert_not_called()
+        assert dag_import_error_listener.new[str(bundle_path / "broken.py")] == persisted[0].stacktrace
+
+    @patch.object(ParseImportError, "full_file_path")
+    @pytest.mark.usefixtures("clean_db")
+    def test_import_error_persisted_without_listener_or_bundle_path(
+        self, mock_full_path, session, testing_dag_bundle
+    ):
+        """
+        With no listener registered, the ``full_file_path()`` fallback must not run at all.
+
+        Callers that do not thread ``bundle_path`` down would otherwise still hit the
+        bundle-instantiating fallback inside the parsing transaction just to build an argument
+        for a hook nobody listens to -- the original silent-drop bug.
+        """
+        mock_full_path.side_effect = AssertionError(
+            "full_file_path() must not be called when no listener is registered"
+        )
+        _update_import_errors(
+            files_parsed={("testing", "abc.py")},
+            bundle_name="testing",
+            import_errors={("testing", "abc.py"): "AnImportError"},
+            session=session,
+        )
+        session.flush()
+
+        persisted = session.scalars(select(ParseImportError)).all()
+        assert len(persisted) == 1
+        assert persisted[0].filename == "abc.py"
+        assert persisted[0].stacktrace == "AnImportError"
+
+        mock_full_path.assert_not_called()
+
     @patch.object(ParseImportError, "full_file_path")
     @mark_fab_auth_manager_test
     @conf_vars({("core", "min_serialized_dag_update_interval"): "5"})

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

@@ -1338,6 +1338,7 @@ def test_handle_parsing_result_updates_stats_after_successful_persist(self, sess
             parsing_result=processor.parsing_result,
             run_duration=mock.ANY,
             relative_fileloc="abc.txt",
+            bundle_path=TEST_DAGS_FOLDER,
             session=session,
         )
         assert manager._file_stats[file] is not original_stat