Emit parse-time DAG warnings for partial dynamic factory failures

airflow-core/docs/migrations-ref.rst

@@ -39,7 +39,10 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
-| ``38c8df3fcb9c`` (head) | ``c71065a1b14f`` | ``3.5.0``         | Add index on dag_run (dag_id, run_after).                    |
+| ``b8f2c1d04e91`` (head) | ``38c8df3fcb9c`` | ``3.5.0``         | Extend dag_warning for parse-time failures (file scope, no   |
+|                         |                  |                   | dag FK).                                                     |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``38c8df3fcb9c``        | ``c71065a1b14f`` | ``3.5.0``         | Add index on dag_run (dag_id, run_after).                    |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``c71065a1b14f``        | ``b4f1a7d2c903`` | ``3.5.0``         | Add session_id and registration_id to Job.                   |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+

airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_warning.py

@@ -20,10 +20,8 @@
 from collections.abc import Iterable
 from datetime import datetime
 
-from pydantic import AliasPath, Field
-
 from airflow.api_fastapi.core_api.base import BaseModel
-from airflow.models.dagwarning import DagWarningTypeValue
+from airflow.models.dagwarning import DagWarning, DagWarningTypeValue
 
 
 class DAGWarningResponse(BaseModel):
@@ -33,7 +31,20 @@ class DAGWarningResponse(BaseModel):
     warning_type: DagWarningTypeValue
     message: str
     timestamp: datetime
-    dag_display_name: str = Field(validation_alias=AliasPath("dag_model", "dag_display_name"))
+    dag_display_name: str
+
+    @classmethod
+    def from_dag_warning(cls, warning: DagWarning) -> DAGWarningResponse:
+        display_name = warning.dag_id
+        if warning.dag_model is not None:
+            display_name = warning.dag_model.dag_display_name
+        return cls(
+            dag_id=warning.dag_id,
+            warning_type=warning.warning_type,
+            message=warning.message,
+            timestamp=warning.timestamp,
+            dag_display_name=display_name,
+        )
 
 
 class DAGWarningCollectionResponse(BaseModel):

airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_warning.py

@@ -38,6 +38,7 @@
 from airflow.api_fastapi.common.router import AirflowRouter
 from airflow.api_fastapi.core_api.datamodels.dag_warning import (
     DAGWarningCollectionResponse,
+    DAGWarningResponse,
 )
 from airflow.api_fastapi.core_api.security import ReadableDagWarningsFilterDep, requires_access_dag
 from airflow.models.dagwarning import DagWarning, DagWarningTypeValue
@@ -66,16 +67,17 @@ def list_dag_warnings(
 ) -> DAGWarningCollectionResponse:
     """Get a list of Dag warnings."""
     dag_warnings_select, total_entries = paginated_select(
-        statement=select(DagWarning).options(joinedload(DagWarning.dag_model)),
+        statement=select(DagWarning),
         filters=[warning_type, dag_id, readable_dag_warning_filter],
         order_by=order_by,
         offset=offset,
         limit=limit,
         session=session,
     )
+    dag_warnings_select = dag_warnings_select.options(joinedload(DagWarning.dag_model))
     dag_warnings = session.scalars(dag_warnings_select)
 
     return DAGWarningCollectionResponse(
-        dag_warnings=dag_warnings,
+        dag_warnings=[DAGWarningResponse.from_dag_warning(w) for w in dag_warnings],
         total_entries=total_entries,
     )

airflow-core/src/airflow/api_fastapi/core_api/security.py

@@ -27,7 +27,7 @@
 from itsdangerous import BadSignature, URLSafeSerializer
 from jwt import ExpiredSignatureError, InvalidTokenError
 from pydantic import NonNegativeInt, TypeAdapter, ValidationError
-from sqlalchemy import or_, select
+from sqlalchemy import and_, or_, select, tuple_
 from sqlalchemy.orm import Session
 
 from airflow.api_fastapi.app import get_auth_manager
@@ -76,7 +76,7 @@
 from airflow.models.dag import DagModel, DagRun, DagTag
 from airflow.models.dag_version import DagVersion
 from airflow.models.dagbundle import DagBundleModel
-from airflow.models.dagwarning import DagWarning
+from airflow.models.dagwarning import PARSE_FAILURE_WARNING_TYPE, DagWarning
 from airflow.models.errors import ParseImportError
 from airflow.models.log import Log
 from airflow.models.taskinstance import TaskInstance as TI
@@ -315,8 +315,24 @@ def to_orm(self, select: Select) -> Select:
 class PermittedDagWarningFilter(PermittedDagFilter):
     """A parameter that filters the permitted dag warnings for the user."""
 
-    def to_orm(self, select: Select) -> Select:
-        return select.where(DagWarning.dag_id.in_(self.value or set()))
+    def to_orm(self, statement: Select) -> Select:
+        authorized = self.value or set()
+        readable_files = (
+            select(DagModel.bundle_name, DagModel.relative_fileloc)
+            .where(DagModel.dag_id.in_(authorized))
+            .distinct()
+        )
+        return statement.where(
+            or_(
+                DagWarning.dag_id.in_(authorized),
+                and_(
+                    DagWarning.warning_type == PARSE_FAILURE_WARNING_TYPE,
+                    DagWarning.bundle_name.is_not(None),
+                    DagWarning.filename.is_not(None),
+                    tuple_(DagWarning.bundle_name, DagWarning.filename).in_(readable_files),
+                ),
+            )
+        )
 
 
 class PermittedEventLogFilter(PermittedDagFilter):

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

@@ -52,7 +52,11 @@
 from airflow.models.dag import DagModel, DagOwnerAttributes, DagTag
 from airflow.models.dagbundle import DagBundleModel
 from airflow.models.dagrun import DagRun
-from airflow.models.dagwarning import DagWarning, DagWarningType
+from airflow.models.dagwarning import (
+    PARSE_FAILURE_WARNING_TYPE,
+    DagWarning,
+    DagWarningType,
+)
 from airflow.models.errors import ParseImportError
 from airflow.models.serialized_dag import SerializedDagModel
 from airflow.models.trigger import Trigger
@@ -383,6 +387,60 @@ def _update_dag_warnings(
         session.merge(warning_to_add)
 
 
+def build_dag_parse_failure_warnings(
+    bundle_name: str, warnings_by_rel_path: dict[str, list]
+) -> list[DagWarning]:
+    return [
+        DagWarning(
+            dag_id=record.dag_id,
+            warning_type=record.warning_type,
+            message=record.message,
+            bundle_name=bundle_name,
+            filename=rel_path,
+        )
+        for rel_path, records in warnings_by_rel_path.items()
+        for record in records
+    ]
+
+
+def _update_file_scoped_parse_failure_warnings(
+    files_parsed: set[tuple[str, str]],
+    parse_failure_warnings: Iterable[DagWarning],
+    session: Session,
+) -> None:
+    if files_parsed:
+        session.execute(
+            delete(DagWarning).where(
+                tuple_(DagWarning.bundle_name, DagWarning.filename).in_(files_parsed),
+                DagWarning.warning_type == PARSE_FAILURE_WARNING_TYPE,
+            )
+        )
+    for warning in parse_failure_warnings:
+        session.merge(warning)
+
+
+def clear_orphaned_file_scoped_parse_failure_warnings(
+    bundle_name: str,
+    observed_filelocs: set[str],
+    session: Session,
+) -> None:
+    """Remove parse-failure warnings for files that no longer exist in the bundle."""
+    try:
+        warnings = session.scalars(
+            select(DagWarning)
+            .where(
+                DagWarning.bundle_name == bundle_name,
+                DagWarning.warning_type == PARSE_FAILURE_WARNING_TYPE,
+            )
+            .options(load_only(DagWarning.filename))
+        )
+        for warning in warnings:
+            if warning.filename not in observed_filelocs:
+                session.delete(warning)
+    except Exception:
+        log.exception("Error removing old parse-failure DAG warnings")
+
+
 def _update_import_errors(
     files_parsed: set[tuple[str, str]],
     import_errors: dict[tuple[str, str], str],
@@ -608,6 +666,7 @@ def update_dag_parsing_results_in_db(
     warnings: set[DagWarning],
     session: Session,
     *,
+    parse_failure_warnings: Iterable[DagWarning] | None = None,
     version_data: dict | None = None,
     warning_types: tuple[DagWarningType, ...] = (
         DagWarningType.DUPLICATE_DAG_ID,
@@ -715,6 +774,15 @@ def update_dag_parsing_results_in_db(
     except Exception:
         log.exception("Error logging DAG warnings.")
 
+    try:
+        _update_file_scoped_parse_failure_warnings(
+            files_parsed if files_parsed is not None else set(),
+            parse_failure_warnings or (),
+            session,
+        )
+    except Exception:
+        log.exception("Error logging Dag parse failure warnings.")
+
     session.flush()
 
 

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

@@ -214,6 +214,7 @@ def __init__(
         self.import_errors: dict[str, str] = {}
         self.captured_warnings: dict[str, tuple[str, ...]] = {}
         self._import_warnings: dict[str, list[DagImportWarning]] = {}
+        self.parse_warnings: dict[str, list] = {}
         self.task_group_cycle_warnings: dict[str, str] = {}
         # The source code of each definition that produced Dags, keyed by its fileloc
         self.dag_source_codes: dict[str, DagSourceCode] = {}
@@ -329,6 +330,8 @@ def _process_definition(
 
         self.captured_warnings.pop(fileloc, None)
         self._import_warnings.pop(fileloc, None)
+        rel_path = self._get_relative_fileloc(fileloc)
+        self.parse_warnings.pop(rel_path, None)
         # Keyed by dag_id, not fileloc: a bundle can embed several Dags' own source under one
         # fileloc, so a stale entry is found by which dag_ids this fileloc previously bagged.
         for dag_id, dag in list(self.dags.items()):
@@ -356,6 +359,12 @@ def _import_sdk_definition(self, importer: AbstractDagImporter, definition: DagD
         for error in result.errors:
             self._record_import_error(error, root=self._bundle.path)
 
+        rel_path = self._get_relative_fileloc(fileloc)
+        if result.parse_warnings:
+            self.parse_warnings[rel_path] = result.parse_warnings
+        else:
+            self.parse_warnings.pop(rel_path, None)
+
         if result.warnings:
             self._import_warnings[fileloc] = result.warnings
             self.captured_warnings[fileloc] = tuple(
@@ -698,6 +707,10 @@ def is_parsed_by_runtime(rel_path: str) -> bool:
             if enclosing_file := find_enclosing_file(Path(dagbag.bundle_path, rel_path)):
                 files_parsed.add((bundle_name, dagbag._get_relative_fileloc(os.fspath(enclosing_file))))
 
+    from airflow.dag_processing.collection import build_dag_parse_failure_warnings
+
+    parse_failure_warnings = build_dag_parse_failure_warnings(bundle_name, dagbag.parse_warnings)
+
     update_dag_parsing_results_in_db(
         bundle_name,
         bundle_version,
@@ -713,4 +726,5 @@ def is_parsed_by_runtime(rel_path: str) -> bool:
         version_data=version_data,
         files_parsed=files_parsed,
         dag_source_codes=dagbag.dag_source_codes,
+        parse_failure_warnings=parse_failure_warnings,
     )

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

@@ -55,7 +55,10 @@
     unpack_bundle_version,
 )
 from airflow.dag_processing.bundles.manager import DagBundlesManager
-from airflow.dag_processing.collection import update_dag_parsing_results_in_db
+from airflow.dag_processing.collection import (
+    clear_orphaned_file_scoped_parse_failure_warnings,
+    update_dag_parsing_results_in_db,
+)
 from airflow.dag_processing.importer_routing import get_claiming_importer
 from airflow.dag_processing.lang_sdk_processor import LangSDKDagFileProcessorProcess
 from airflow.dag_processing.processor import (
@@ -989,6 +992,10 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]):
                 bundle_name=bundle.name,
                 observed_filelocs=self._get_observed_filelocs(found_files),
             )
+            self.clear_orphaned_parse_failure_warnings(
+                bundle_name=bundle.name,
+                observed_filelocs=self._get_observed_filelocs(found_files),
+            )
 
         if any_refreshed:
             # Bundle-to-team assignments can only change on bundle refresh, so clear the cache.
@@ -1101,6 +1108,14 @@ def clear_orphaned_import_errors(
         except Exception:
             self.log.exception("Error removing old import errors")
 
+    @provide_session
+    def clear_orphaned_parse_failure_warnings(
+        self, bundle_name: str, observed_filelocs: set[str], *, session: Session = NEW_SESSION
+    ):
+        """Clear parse-failure warnings for files that no longer exist in the bundle."""
+        self.log.debug("Removing old parse-failure DAG warnings")
+        clear_orphaned_file_scoped_parse_failure_warnings(bundle_name, observed_filelocs, session=session)
+
     def _log_file_processing_stats(self, known_files: dict[str, set[DagFileInfo]]):
         """
         Print out stats about how files are getting processed.
@@ -1386,6 +1401,8 @@ def persist_parsing_result(
         if warnings and isinstance(warnings[0], dict):
             warnings = [DagWarning(**warn) for warn in warnings]
 
+        parse_failure_warnings = [DagWarning(**row) for row in (parsing_result.parse_warnings or [])]
+
         update_dag_parsing_results_in_db(
             bundle_name=bundle_name,
             bundle_version=bundle_version,
@@ -1397,6 +1414,7 @@ def persist_parsing_result(
             session=session,
             files_parsed=files_parsed,
             dag_source_codes=parsing_result.dag_source_codes,
+            parse_failure_warnings=parse_failure_warnings,
         )
 
     def _collect_results(self):

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

@@ -134,6 +134,7 @@ class DagFileParsingResult(BaseModel):
     import_errors: dict[str, str] | None = None
     parsed_definitions: list[str] = Field(default_factory=list)
     """Bundle-relative locations of the Dag definitions imported from ``fileloc``."""
+    parse_warnings: list[dict] = Field(default_factory=list)
     dag_source_codes: dict[str, DagSourceCode] = Field(default_factory=dict)
     """Source code of the parsed Dags, keyed by dag_id."""
     type: Literal["DagFileParsingResult"] = "DagFileParsingResult"
@@ -259,12 +260,24 @@ def _parse_file(msg: DagFileParseRequest, log: FilteringBoundLogger) -> DagFileP
 
     serialized_dags, serialization_import_errors = _serialize_dags(bag, log)
     bag.import_errors.update(serialization_import_errors)
-    # Dag warnings reference the dag table, so only Dags that serialized can carry one.
+    # Pool/runtime warnings reference the dag table; parse-failure warnings are stored separately.
     serialized_dag_ids = {dag.dag_id for dag in serialized_dags}
+    parse_warnings = [
+        {
+            "bundle_name": msg.bundle_name,
+            "filename": rel_path,
+            "dag_id": w.dag_id,
+            "warning_type": w.warning_type,
+            "message": w.message,
+        }
+        for rel_path, warnings in bag.parse_warnings.items()
+        for w in warnings
+    ]
     result = DagFileParsingResult(
         fileloc=msg.file,
         serialized_dags=serialized_dags,
         import_errors=bag.import_errors,
+        parse_warnings=parse_warnings,
         warnings=[
             *stability_check_result.get_formatted_warnings(bag.dag_ids),
             *(