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

[kjh0623]

_update_import_errors() eagerly evaluated ParseImportError.full_file_path() to build the on_new/on_existing_dag_import_error listener arguments. full_file_path() instantiates the DAG bundle via DagBundlesManager().get_bundle(), which for some bundle types (e.g. the Git bundle) resolves an Airflow Connection. Performing that work inside the DAG-parsing transaction can disturb it and roll back the not-yet-flushed ParseImportError row, so the DAG ends up flagged as broken (has_import_errors) while no import error is ever shown in the UI.

Fix

• Thread the already-known bundle path (DagFileInfo.bundle_path) down through persist_parsing_result → update_dag_parsing_results_in_db → _update_import_errors, and build the listener filename with str(bundle_path / relative_fileloc) instead of calling full_file_path(). The bundle is no longer re-instantiated inside the transaction, regardless of which listeners are registered.
• sync_bag_to_db() (used by the CLI and dag.test()) now forwards dagbag.bundle_path as well, so that caller is covered too (review feedback from @Vamsi-klu).
• When no listener plugin is registered at all, the listener arguments are not computed in the first place (cheap has_listeners early exit), so callers that don't thread bundle_path down never hit the full_file_path() fallback inside the transaction.
• As a side benefit, the existing-error path no longer needs the extra select(ParseImportError) query that only existed to feed the listener argument (suggested by @dhkim1920).

Tests

• Regression test through sync_bag_to_db() end-to-end: full_file_path() raises if called, so the test fails if any production caller stops forwarding bundle_path.
• Coverage for the on_existing_dag_import_error branch with bundle_path.
• Coverage for the no-listener case without bundle_path (the original silent-drop scenario).

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[Vamsi-klu]

Had a look through this and the guard reads right to me. full_file_path() in models/errors.py does DagBundlesManager().get_bundle(...).path, so gating both notification calls in _update_import_errors behind get_listener_manager().has_listeners keeps that bundle instantiation out of the parsing transaction when no listener is registered, which is the common case. Nice that it reuses the same has_listeners check already in dagbag.py instead of inventing a new signal. The regression test mocking full_file_path with assert_not_called() and then asserting the ParseImportError row still persists lands right on the behavior that was broken, and as a bonus the existing-error branch now skips the extra select(ParseImportError) query that only ever existed to feed the listener arg.

[kjh0623]

Thanks for the review!

[dhkim1920]

Thanks for fixing the no-listener case.

I think there may still be a similar transaction issue when any listener plugin is registered.

ListenerManager.has_listeners only checks whether any listener plugin exists, not whether an import-error hook is actually implemented. That means ParseImportError.full_file_path() is still evaluated while building the hook arguments, even if none of the registered listeners handle import errors.

For bundle types like the Git bundle, calling full_file_path() can instantiate the bundle and resolve an Airflow Connection inside the parsing transaction. If that ends up rolling back the session, the pending import-error row could still be lost.

Would it be reasonable to pass the already-known bundle_path down the parsing call chain and build the listener filename with str(bundle_path / relative_fileloc) instead? DagFileInfo and DagBag already have that information, so it avoids initializing the bundle again inside the transaction regardless of which listeners are registered.

As a side benefit, the existing-error path would no longer need the extra select(ParseImportError) query just to construct the listener filename.

I tried this approach locally with a registered listener for both new and existing import errors, and the listener still received the expected filename while the ParseImportError row remained persisted without calling full_file_path().

[kjh0623]

Thanks @dhkim1920, that's a great catch — you're right that has_listeners only tells us whether any listener plugin is registered, not whether an import-error hook is actually implemented, so full_file_path() (and the bundle instantiation it triggers) could still run inside the parsing transaction whenever any listener exists.

I've updated the PR to your suggested approach. The already-known bundle path (DagFileInfo.bundle_path) is now threaded down through persist_parsing_result → update_dag_parsing_results_in_db → _update_import_errors, and the listener filename is built with str(bundle_path / relative_fileloc). So the bundle is no longer re-instantiated inside the transaction regardless of which listeners are registered, and — as you noted — the existing-error path no longer needs the extra select(ParseImportError) that only existed to feed the listener argument. full_file_path() is kept only as a fallback for callers that don't thread the bundle path.

The regression test now registers an import-error listener and asserts that full_file_path() is not called while the listener still receives the correct bundle_path-based filename and the ParseImportError row is persisted.

@Vamsi-klu thanks for the earlier review as well — this approach makes the has_listeners guard unnecessary, so I've dropped it in favor of threading bundle_path.

[Vamsi-klu]

I rechecked the updated bundle-path implementation. The new direction is better, but one production caller still does not forward the path. sync_bag_to_db() already has dagbag.bundle_path and uses it while constructing files_parsed, but its call to update_dag_parsing_results_in_db() omits bundle_path. That path can therefore still fall back to ParseImportError.full_file_path() inside the transaction.

Could you pass bundle_path=dagbag.bundle_path there as well and add a regression test through sync_bag_to_db() that makes full_file_path() fail if called? The current test invokes the private _update_import_errors() helper directly with a supplied path, so it stays green if a production caller stops forwarding that path. Coverage for an existing import-error row would also protect the other changed listener branch.

The title and body still describe the superseded has_listeners implementation, and the AI disclosure does not name the tool, so those should be refreshed before rereview. Only the WIP and Mergeable gates have run on this head; substantive CI is still needed.

---

Drafted-by: Codex (GPT-5); reviewed by @Vamsi-klu before posting

[kjh0623]

Thanks @Vamsi-klu for the thorough recheck — all points addressed in 364a7d0:

sync_bag_to_db() now forwards the path: it passes bundle_path=dagbag.bundle_path to update_dag_parsing_results_in_db(), so the CLI / dag.test() path no longer falls back to full_file_path() inside the transaction.
End-to-end regression test: added test_sync_bag_to_db_does_not_call_full_file_path, which builds a real DagBag over a broken DAG file and goes through sync_bag_to_db() (not the private helper). full_file_path is patched with side_effect=AssertionError, so the test fails immediately if any production caller stops forwarding bundle_path.
Existing-error branch coverage: added test_existing_import_error_listener_uses_bundle_path_without_full_file_path, which pre-creates a ParseImportError row and asserts on_existing_dag_import_error receives the bundle_path-based filename without hitting full_file_path() or the extra select.
Belt and braces: while revisiting this I also restored a cheap has_listeners early exit in _update_import_errors. Threading bundle_path covers the known callers, but any caller that doesn't thread it down would still have hit the bundle-instantiating fallback in the no-listener case — the original bug. With the gate, the listener arguments aren't computed at all when no listener plugin is registered (covered by test_import_error_persisted_without_listener_or_bundle_path).
Title / body / AI disclosure: refreshed to describe the current bundle_path implementation, and the disclosure now names the tool.
Ready for CI whenever a maintainer can approve the workflows.