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

[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]

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.