[ydsakshi]

Description

Addresses #29275.

Partial failures in dynamic DAG factories/files can call emit_dag_warning() during file parse. The DAG processor takes those warnings on DagFileParsingResult & stores them in dag_warning table (with columns - bundle_name & filename) & exposes them via GET /api/v2/dagWarnings & a Dashboard parse-failure card.

Reproduced the issue using the below sample script on main & then verified the fix on feature branch.

Evidence

Sample dynamic DAG creator file :-

from __future__ import annotations

import logging
from datetime import datetime
import traceback


from airflow.sdk import dag, task
#from airflow.sdk.definitions.parse_warning import emit_dag_warning

log = logging.getLogger(__name__)

FAILING_DAG_IDS = {"B", "E"}

for dag_id in "ABCDE":
    try:
        if dag_id in FAILING_DAG_IDS:
            raise RuntimeError(f"Failed to create dynamic DAG {dag_id}")

        @dag(dag_id=f"repro_29275_{dag_id}", schedule=None, start_date=datetime(2023, 1, 31), catchup=False)
        def dynamic_dag():
            @task
            def t():
                print(f"inside task {dag_id}")

            t()

        dynamic_dag()
    except Exception:
        #emit_dag_warning(
        #    f"repro_{dag_id}", f"Failed to create dynamic DAG {dag_id}", stacktrace=traceback.format_exc()
        #)
        log.exception("Failed to create dynamic DAG %s", dag_id)

Scenario	Evidence
Correct Dags(A,C,D) Load (main)
Correct Dags(A,C,D) Load (feature)
__	__
Dashboard UI(main) - No warnings for failing DAGs(B,E)
Dashboard UI(feature) - Proper warnings(with exceptions) for failing DAGs(B,E)	,

---

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

• Yes (please specify the tool below)

Not generated by AI, used Cursor for approach & design finalization. Code is fully vetted by me.

---

• 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.

[ydsakshi]

Local testing & Evidences attached ✅
Self Review ✅
Local Pre-commit checks ✅
Have highlighted auto generated files.

[FrankYang0529]

Thanks for the PR. Leave some comments.

[shahar1]

Thanks for your contribution!
The idea looks solid overall, the PR should be polished a bit more.
Besides, I think it's worth to post a discussion followed by lazy consensus on the dev list, since it involves migrations and schema changes (I wouldn't write a full AIP for that).

@uranusjr - you might want to take a look as well at some point.

[shahar1]

1. emit_dag_warning() accepts a custom warning_type, but the cleanup functions, the access filter and the Dashboard card only look at PARSE_FAILURE_WARNING_TYPE`.

2. The cleanup for inactive Dags also skips
   these rows, so a warning with a custom type stays in the table forever.

[shahar1]

Anyone who can read one Dag in the file sees the stacktrace. For import errors from the same file, the API hides
the stacktrace unless the user can read all Dags in it. If no Dag in the file loaded, nobody can see the warning.

[shahar1]

Could you please add a docstring? (otherwise description below is None)

[shahar1]

Could you extract the import to the top-level of this module?

[shahar1]

Every parse now runs an extra DELETE on (bundle_name, filename), and neither column is indexed. The PR raised the
statement counts in test_manager.py to fit it.

[shahar1]

There are no tests for task-sdk at all - this one needs to be well-covered, since it's the essence of this PR.

[shahar1]

Could you please rename the function so it would start with a verb?

[shahar1]

nit: session should be keyword-only argument

Suggested change

	def _update_file_scoped_parse_failure_warnings(
	files_parsed: set[tuple[str, str]],
	parse_failure_warnings: Iterable[DagWarning],
	session: Session,
	) -> None:
	def _update_file_scoped_parse_failure_warnings(
	*,
	files_parsed: set[tuple[str, str]],
	parse_failure_warnings: Iterable[DagWarning],
	session: Session,
	) -> None:

[shahar1]

You repeat the literal user:dynamic_dag_failure few times in the code rather importing it from a single source of truth, please consider changing that where it's not necessary to repeat.