From b679d3dd8d7b49d6690f6412f5be3f2cf0a54647 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Sun, 25 Oct 2026 21:58:40 +0530 Subject: [PATCH 01/21] initial change files --- .../core_api/datamodels/dag_warning.py | 13 +++- .../openapi/v2-rest-api-generated.yaml | 5 +- .../airflow/api_fastapi/core_api/security.py | 22 +++++- .../src/airflow/dag_processing/collection.py | 73 ++++++++++++------- .../src/airflow/dag_processing/dagbag.py | 14 ++++ .../dag_processing/lang_sdk_processor.py | 4 +- .../src/airflow/dag_processing/manager.py | 5 +- .../src/airflow/dag_processing/processor.py | 35 +++++---- .../example_ext_task_sensor_child.py | 46 ++++++++++++ .../example_ext_task_sensor_parent.py | 33 +++++++++ .../0150_3_5_0_add_dag_parse_warning.py | 45 ++++++++++++ airflow-core/src/airflow/models/dagwarning.py | 39 ++++++---- .../ui/public/i18n/locales/en/dashboard.json | 4 +- .../Dashboard/Stats/ParseFailureWarnings.tsx | 68 +++++++++++++++++ .../ui/src/pages/Dashboard/Stats/Stats.tsx | 3 + .../routes/public/test_dag_warning.py | 21 +++++- .../tests/unit/models/test_dagwarning.py | 24 +++++- .../example_dags/example_repro_29275.py | 29 ++++++++ task-sdk/src/airflow/sdk/__init__.py | 5 +- .../definitions/_internal/contextmanager.py | 2 + .../airflow/sdk/definitions/parse_warning.py | 55 ++++++++++++++ .../sdk/definitions/parse_warning_record.py | 10 +++ task-sdk/src/airflow/sdk/importers/base.py | 2 + .../airflow/sdk/importers/python_importer.py | 13 ++-- 24 files changed, 492 insertions(+), 78 deletions(-) create mode 100644 airflow-core/src/airflow/example_dags/example_ext_task_sensor_child.py create mode 100644 airflow-core/src/airflow/example_dags/example_ext_task_sensor_parent.py create mode 100644 airflow-core/src/airflow/migrations/versions/0150_3_5_0_add_dag_parse_warning.py create mode 100644 airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx create mode 100644 providers/standard/src/airflow/providers/standard/example_dags/example_repro_29275.py create mode 100644 task-sdk/src/airflow/sdk/definitions/parse_warning.py create mode 100644 task-sdk/src/airflow/sdk/definitions/parse_warning_record.py diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_warning.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_warning.py index 24dd095c75a06..e49125dd87b99 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_warning.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_warning.py @@ -20,7 +20,7 @@ from collections.abc import Iterable from datetime import datetime -from pydantic import AliasPath, Field +from pydantic import AliasPath, Field, model_validator from airflow.api_fastapi.core_api.base import BaseModel from airflow.models.dagwarning import DagWarningTypeValue @@ -33,7 +33,16 @@ 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 | None = Field( + default=None, validation_alias=AliasPath("dag_model", "dag_display_name") + ) + + +@model_validator(mode="after") +def default_dag_display_name(self) -> DAGWarningResponse: + if not self.dag_display_name: + self.dag_display_name = self.dag_id + return self class DAGWarningCollectionResponse(BaseModel): diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml index aecd0bdeaf05a..2b34c54259fca 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml @@ -15556,7 +15556,9 @@ components: format: date-time title: Timestamp dag_display_name: - type: string + anyOf: + - type: string + - type: 'null' title: Dag Display Name type: object required: @@ -15564,7 +15566,6 @@ components: - warning_type - message - timestamp - - dag_display_name title: DAGWarningResponse description: Dag Warning serializer for responses. DagBundleCollectionResponse: diff --git a/airflow-core/src/airflow/api_fastapi/core_api/security.py b/airflow-core/src/airflow/api_fastapi/core_api/security.py index bf2821e972e75..edb4f289291a7 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/security.py +++ b/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 @@ -316,7 +316,23 @@ 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())) + authorized = self.value or set() + readable_files = ( + select(DagModel.bundle_name, DagModel.relative_fileloc) + .where(DagModel.dag_id.in_(authorized)) + .distinct() + ) + return select.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): diff --git a/airflow-core/src/airflow/dag_processing/collection.py b/airflow-core/src/airflow/dag_processing/collection.py index 7045189005092..9ed35f252be82 100644 --- a/airflow-core/src/airflow/dag_processing/collection.py +++ b/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,36 @@ def _update_dag_warnings( session.merge(warning_to_add) +def 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 _update_import_errors( files_parsed: set[tuple[str, str]], import_errors: dict[tuple[str, str], str], @@ -573,32 +607,6 @@ def _reject_other_teams_plugin_classes( return accepted -def _reject_dags_with_unloadable_timetables( - bundle_name: str, - dags: Collection[LazyDeserializedDAG], - import_errors: dict[tuple[str, str], str], -) -> list[LazyDeserializedDAG]: - """ - Drop Dags whose timetable raises when rebuilt, recording an import error for each. - - A custom timetable runs user code when it is deserialized, for example ``Variable.get`` in - ``__init__``. Checking here keeps one failing Dag from aborting the save of every Dag in its file. - """ - accepted: list[LazyDeserializedDAG] = [] - for dag in dags: - try: - dag.timetable - except Exception: - log.exception("Failed to load timetable of Dag", dag_id=dag.dag_id) - depth = conf.getint("core", "dagbag_import_error_traceback_depth") - import_errors[(bundle_name, dag.relative_fileloc)] = ( - f"Failed to load the timetable of Dag '{dag.dag_id}':\n{traceback.format_exc(limit=-depth)}" - ) - continue - accepted.append(dag) - return accepted - - def update_dag_parsing_results_in_db( bundle_name: str, bundle_version: str | None, @@ -608,6 +616,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, @@ -642,7 +651,6 @@ def update_dag_parsing_results_in_db( without an entry have their source read from ``fileloc``. """ accepted = _reject_other_teams_plugin_classes(bundle_name, dags, import_errors, session=session) - accepted = _reject_dags_with_unloadable_timetables(bundle_name, accepted, import_errors) if len(accepted) != len(dags): # A rejected Dag may have no ``dag`` row yet, and dag_warning has a foreign key to it. rejected_ids = {dag.dag_id for dag in dags} - {dag.dag_id for dag in accepted} @@ -715,6 +723,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() diff --git a/airflow-core/src/airflow/dag_processing/dagbag.py b/airflow-core/src/airflow/dag_processing/dagbag.py index dc80433676ca7..23943c923f176 100644 --- a/airflow-core/src/airflow/dag_processing/dagbag.py +++ b/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 dag_parse_failure_warnings + + parse_failure_warnings = 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, ) diff --git a/airflow-core/src/airflow/dag_processing/lang_sdk_processor.py b/airflow-core/src/airflow/dag_processing/lang_sdk_processor.py index 9e753e61059c1..41ecf8b66a65b 100644 --- a/airflow-core/src/airflow/dag_processing/lang_sdk_processor.py +++ b/airflow-core/src/airflow/dag_processing/lang_sdk_processor.py @@ -42,7 +42,7 @@ BaseDagFileProcessorProcess, DagFileParseRequest, DagFileParsingResult, - ToManagerMessage, + ToManager, ) from airflow.exceptions import DeserializationError from airflow.models.pool import Pool @@ -101,7 +101,7 @@ class LangSDKRuntimeSchemaVersion(BaseModel): type: Literal["LangSDKRuntimeSchemaVersion"] = "LangSDKRuntimeSchemaVersion" -LangSDKToManager = Annotated[LangSDKRuntimeSchemaVersion | ToManagerMessage, Field(discriminator="type")] +LangSDKToManager = Annotated[LangSDKRuntimeSchemaVersion | ToManager, Field(discriminator="type")] def _get_import_timeout(path: str) -> float | None: diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index a4a834b9acec1..04cc765e90c7f 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/airflow-core/src/airflow/dag_processing/manager.py @@ -1392,6 +1392,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, @@ -1403,6 +1405,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): @@ -1868,7 +1871,7 @@ def process_parse_results( normalized_bundle = normalize_name_for_stats(bundle_name) stats.timing( "dag_processing.last_duration", - timedelta(seconds=stat.last_duration), + stat.last_duration, tags=prune_dict( {"bundle_name": normalized_bundle, "file_name": file_name, "team_name": team_name} ), diff --git a/airflow-core/src/airflow/dag_processing/processor.py b/airflow-core/src/airflow/dag_processing/processor.py index 4be5623d71dd9..76221cd88908f 100644 --- a/airflow-core/src/airflow/dag_processing/processor.py +++ b/airflow-core/src/airflow/dag_processing/processor.py @@ -73,11 +73,7 @@ XComSequenceSliceResult, ) from airflow.sdk.execution_time.supervisor import WatchedSubprocess, register_request_method -from airflow.sdk.execution_time.task_runner import ( - RuntimeTaskInstance, - _send_error_email_notification, - check_xcom_backend_supports_loops, -) +from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance, _send_error_email_notification from airflow.sdk.importers import DagSourceCode # noqa: TC001 from airflow.serialization.serialized_objects import DagSerialization, LazyDeserializedDAG from airflow.utils.dag_version_inflation_checker import check_dag_file_stability @@ -134,12 +130,13 @@ 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" -ToManagerMessage = ( +ToManager = Annotated[ DagFileParsingResult | GetConnection | GetVariable @@ -155,10 +152,9 @@ class DagFileParsingResult(BaseModel): | GetXComCount | GetXComSequenceItem | GetXComSequenceSlice - | MaskSecret -) - -ToManager = Annotated[ToManagerMessage, Field(discriminator="type")] + | MaskSecret, + Field(discriminator="type"), +] ToDagProcessor = Annotated[ DagFileParseRequest @@ -259,12 +255,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), *( @@ -287,9 +295,6 @@ def _serialize_dags( serialized_dags = [] for dag in bag.dags.values(): try: - # Only possible because importing the SDK resolves the XCom backend here; the worker check is the - # one that must stay, so drop this one rather than keep that import working for its sake. - check_xcom_backend_supports_loops(dag) data = DagSerialization.to_dict(dag) serialized_dags.append(LazyDeserializedDAG(data=data, last_loaded=dag.last_loaded)) except Exception: @@ -594,6 +599,7 @@ class BaseDagFileProcessorProcess(WatchedSubprocess, LoggingMixin): ) """The file's parse log, which the process closes. Without one, the output is discarded.""" parsing_result: DagFileParsingResult | None = None + decoder: ClassVar[TypeAdapter[ToManager]] = TypeAdapter[ToManager](ToManager) had_callbacks: bool = False # Track if this process was started with callbacks to prevent stale DAG detection false positives client: Client | None = None @@ -718,7 +724,6 @@ class DagFileProcessorProcess(BaseDagFileProcessorProcess): """ client: Client - decoder: ClassVar[TypeAdapter[ToManager]] = TypeAdapter[ToManager](ToManager) _request_handlers: ClassVar[dict[type[BaseModel], RequestHandler[DagFileProcessorProcess]]] = { **BaseDagFileProcessorProcess._common_request_handlers, diff --git a/airflow-core/src/airflow/example_dags/example_ext_task_sensor_child.py b/airflow-core/src/airflow/example_dags/example_ext_task_sensor_child.py new file mode 100644 index 0000000000000..1d12096bc2cdd --- /dev/null +++ b/airflow-core/src/airflow/example_dags/example_ext_task_sensor_child.py @@ -0,0 +1,46 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from __future__ import annotations + +from datetime import datetime, timedelta + +from airflow.sdk import DAG +from airflow.sensors.external_task import ExternalTaskSensor + + +def get_previous_execution_date(logical_date, **kwargs): + """Get the previous execution date of the parent DAG.""" + return logical_date - timedelta(minutes=5) + + +with DAG( + dag_id="example_ext_task_sensor_child", + start_date=datetime(2026, 2, 11), + schedule="42 00 * * *", + catchup=False, + tags=["example", "ext-sensor"], + description="This DAG is used to demonstrate the ExternalTaskSensor. It is used in conjunction with example_ext_task_sensor_parent.py", +) as dag: + wait_for_parent = ExternalTaskSensor( + task_id="wait_for_parent", + external_dag_id="example_ext_task_sensor_parent", + execution_date_fn=get_previous_execution_date, + poke_interval=30, + mode="poke", + timeout=300, + ) diff --git a/airflow-core/src/airflow/example_dags/example_ext_task_sensor_parent.py b/airflow-core/src/airflow/example_dags/example_ext_task_sensor_parent.py new file mode 100644 index 0000000000000..c8dd56822c573 --- /dev/null +++ b/airflow-core/src/airflow/example_dags/example_ext_task_sensor_parent.py @@ -0,0 +1,33 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from __future__ import annotations + +from datetime import datetime + +from airflow.operators.empty import EmptyOperator +from airflow.sdk import DAG + +with DAG( + dag_id="example_ext_task_sensor_parent", + start_date=datetime(2026, 2, 11), + schedule="37 00 * * *", + catchup=False, + tags=["example", "ext-sensor"], + description="This DAG is used to demonstrate the ExternalTaskSensor. It is used in conjunction with example_ext_task_sensor_child.py", +) as dag: + EmptyOperator(task_id="parent_task") diff --git a/airflow-core/src/airflow/migrations/versions/0150_3_5_0_add_dag_parse_warning.py b/airflow-core/src/airflow/migrations/versions/0150_3_5_0_add_dag_parse_warning.py new file mode 100644 index 0000000000000..efd7a0e0ad0a0 --- /dev/null +++ b/airflow-core/src/airflow/migrations/versions/0150_3_5_0_add_dag_parse_warning.py @@ -0,0 +1,45 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Extend dag_warning for parse-time failures (file scope, no dag FK).""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +from airflow.models.base import StringID + +revision = "b8f2c1d04e91" +down_revision = "c71065a1b14f" +branch_labels = None +depends_on = None +airflow_version = "3.4.0" + + +def upgrade(): + with op.batch_alter_table("dag_warning") as batch_op: + batch_op.drop_constraint("dcw_dag_id_fkey", type_="foreignkey") + batch_op.add_column(sa.Column("bundle_name", StringID(), nullable=True)) + batch_op.add_column(sa.Column("filename", sa.String(length=1024), nullable=True)) + + +def downgrade(): + with op.batch_alter_table("dag_warning") as batch_op: + batch_op.drop_column("filename") + batch_op.drop_column("bundle_name") + batch_op.create_foreign_key("dcw_dag_id_fkey", "dag", ["dag_id"], ["dag_id"], ondelete="CASCADE") diff --git a/airflow-core/src/airflow/models/dagwarning.py b/airflow-core/src/airflow/models/dagwarning.py index 4d1eeac7db59d..a1ffbfda4cedf 100644 --- a/airflow-core/src/airflow/models/dagwarning.py +++ b/airflow-core/src/airflow/models/dagwarning.py @@ -22,7 +22,7 @@ from typing import TYPE_CHECKING, Annotated from pydantic import StringConstraints, TypeAdapter -from sqlalchemy import ForeignKeyConstraint, Index, String, Text, delete, select, true +from sqlalchemy import Index, String, Text, delete, select, true from sqlalchemy.orm import Mapped, mapped_column, relationship from airflow._shared.timezones import timezone @@ -36,6 +36,7 @@ from sqlalchemy.orm import Session WARNING_TYPE_MAX_LENGTH = 50 +PARSE_FAILURE_WARNING_TYPE = "user:dynamic_dag_failure" class DagWarning(Base): @@ -51,25 +52,35 @@ class DagWarning(Base): warning_type: Mapped[str] = mapped_column(String(WARNING_TYPE_MAX_LENGTH), primary_key=True) message: Mapped[str] = mapped_column(Text, nullable=False) timestamp: Mapped[datetime] = mapped_column(UtcDateTime, nullable=False, default=timezone.utcnow) - - dag_model = relationship("DagModel", viewonly=True, lazy="selectin") - - __tablename__ = "dag_warning" - __table_args__ = ( - ForeignKeyConstraint( - ("dag_id",), - ["dag.dag_id"], - name="dcw_dag_id_fkey", - ondelete="CASCADE", - ), - Index("idx_dag_warning_dag_id", dag_id), + bundle_name: Mapped[str | None] = mapped_column(StringID(), nullable=True) + filename: Mapped[str | None] = mapped_column(String(1024), nullable=True) + + dag_model = relationship( + "DagModel", + viewonly=True, + lazy="selectin", + primaryjoin="foreign(DagWarning.dag_id) == DagModel.dag_id", ) - def __init__(self, dag_id: str, warning_type: str, message: str, **kwargs): + __tablename__ = "dag_warning" + __table_args__ = (Index("idx_dag_warning_dag_id", "dag_id"),) + + def __init__( + self, + dag_id: str, + warning_type: str, + message: str, + *, + bundle_name: str | None = None, + filename: str | None = None, + **kwargs, + ): super().__init__(**kwargs) self.dag_id = dag_id self.warning_type = get_warning_type_value(warning_type) self.message = message + self.bundle_name = bundle_name + self.filename = filename def __eq__(self, other) -> bool: return self.dag_id == other.dag_id and self.warning_type == other.warning_type diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/dashboard.json b/airflow-core/src/airflow/ui/public/i18n/locales/en/dashboard.json index 698b017bf8dcc..2fa134fef9bcc 100644 --- a/airflow-core/src/airflow/ui/public/i18n/locales/en/dashboard.json +++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/dashboard.json @@ -74,8 +74,10 @@ "queuedDags": "Queued Dags", "requiredActions": "Required Actions", "runningDags": "Running Dags", + "parseFailureWarnings_one": "Dag load warning", + "parseFailureWarnings_other": "Dag load warnings", "stats": "Stats" }, "uri": "Uri", "welcome": "Welcome" -} +} \ No newline at end of file diff --git a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx new file mode 100644 index 0000000000000..b5dd815176844 --- /dev/null +++ b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx @@ -0,0 +1,68 @@ +/*! + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +import { useDisclosure } from "@chakra-ui/react"; +import { useTranslation } from "react-i18next"; +import { LuFileWarning } from "react-icons/lu"; + +import { useDagWarningServiceListDagWarnings } from "openapi/queries"; + +import { DAGWarningsModal } from "src/components/DAGWarningsModal"; +import { StatsCard } from "src/components/StatsCard"; + +import { useAutoRefresh } from "src/utils"; + +const PARSE_FAILURE_WARNING_TYPE = "user:dynamic_dag_failure"; + +export const ParseFailureWarnings = () => { + const { onClose, onOpen, open } = useDisclosure(); + const refetchInterval = useAutoRefresh({ checkPendingRuns: true }); + const { i18n, t: translate } = useTranslation("dashboard"); + const { data, isLoading } = useDagWarningServiceListDagWarnings( + { limit: 100, warningType: PARSE_FAILURE_WARNING_TYPE }, + undefined, + { refetchInterval }, + ); + const count = data?.total_entries ?? 0; + + if (!isLoading && count === 0) { + return undefined; + } + + return ( + <> + } + isLoading={isLoading} + isRTL={i18n.dir() === "rtl"} + label={translate("stats.parseFailureWarnings", { count })} + onClick={onOpen} + /> + ({ + ...warning, + message: `${warning.dag_id}: ${warning.message}`, + }))} + /> + + ); +}; diff --git a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/Stats.tsx b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/Stats.tsx index 612dc1d2f73f1..362bdf54a26a1 100644 --- a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/Stats.tsx +++ b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/Stats.tsx @@ -28,6 +28,7 @@ import { StatsCard } from "src/components/StatsCard"; import { useAutoRefresh } from "src/utils"; import { DagImportErrors } from "./DagImportErrors"; +import { ParseFailureWarnings } from "./ParseFailureWarnings"; import { PluginImportErrors } from "./PluginImportErrors"; export const Stats = () => { @@ -68,6 +69,8 @@ export const Stats = () => { + + {queuedDagsCount > 0 ? ( diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py index e96311a666c67..4e42e47d3a64b 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py @@ -21,7 +21,7 @@ from sqlalchemy.orm import Session from airflow.models.dag import DagModel -from airflow.models.dagwarning import DagWarning +from airflow.models.dagwarning import PARSE_FAILURE_WARNING_TYPE, DagWarning from airflow.utils.session import NEW_SESSION, provide_session from tests_common.test_utils.asserts import assert_queries_count @@ -138,3 +138,22 @@ def test_get_dag_warnings_importer_defined_type(self, test_client, session): assert response_json["total_entries"] == 1 assert response_json["dag_warnings"][0]["warning_type"] == "yaml:schema_violation" assert response_json["dag_warnings"][0]["message"] == "importer message" + + def test_get_parse_failure_warning_without_dag_row(self, test_client, session): + dag1 = session.get(DagModel, DAG1_ID) + dag1.relative_fileloc = "dynamic.py" + session.add( + DagWarning( + "orphan_dag", + PARSE_FAILURE_WARNING_TYPE, + "factory failed", + bundle_name="testing", + filename="dynamic.py", + ) + ) + session.commit() + response = test_client.get("/dagWarnings", params={"warning_type": PARSE_FAILURE_WARNING_TYPE}) + assert response.status_code == 200 + row = response.json()["dag_warnings"][0] + assert row["dag_id"] == "orphan_dag" + assert row["dag_display_name"] == "orphan_dag" diff --git a/airflow-core/tests/unit/models/test_dagwarning.py b/airflow-core/tests/unit/models/test_dagwarning.py index 0f07d7a91924b..c0675cdc68364 100644 --- a/airflow-core/tests/unit/models/test_dagwarning.py +++ b/airflow-core/tests/unit/models/test_dagwarning.py @@ -25,7 +25,13 @@ from sqlalchemy.exc import OperationalError from airflow.models import DagModel -from airflow.models.dagwarning import DagWarning, DagWarningType, get_warning_type_value +from airflow.dag_processing.collection import _update_file_scoped_parse_failure_warnings +from airflow.models.dagwarning import ( + PARSE_FAILURE_WARNING_TYPE, + DagWarning, + DagWarningType, + get_warning_type_value, +) from tests_common.test_utils.db import clear_db_dags @@ -107,3 +113,19 @@ def test_retry_purge_inactive_dag_warnings(self, delete_mock): # Assert that the delete method was called twice assert delete_mock.call_count == 2 assert self.session_mock.execute.call_count == 2 + + def test_file_scoped_parse_failure_warnings_replace_per_file(self, session): + files = {("testing", "dynamic.py")} + warning = DagWarning( + "failed_dag", + PARSE_FAILURE_WARNING_TYPE, + "boom", + bundle_name="testing", + filename="dynamic.py", + ) + _update_file_scoped_parse_failure_warnings(files, [warning], session) + session.commit() + assert len(session.scalars(select(DagWarning)).all()) == 1 + _update_file_scoped_parse_failure_warnings(files, [], session) + session.commit() + assert session.scalars(select(DagWarning)).all() == [] diff --git a/providers/standard/src/airflow/providers/standard/example_dags/example_repro_29275.py b/providers/standard/src/airflow/providers/standard/example_dags/example_repro_29275.py new file mode 100644 index 0000000000000..f199cfc693047 --- /dev/null +++ b/providers/standard/src/airflow/providers/standard/example_dags/example_repro_29275.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +import logging +from datetime import datetime + + +from airflow.sdk import dag, task, 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=None) + log.exception("Failed to create dynamic DAG %s", dag_id) diff --git a/task-sdk/src/airflow/sdk/__init__.py b/task-sdk/src/airflow/sdk/__init__.py index 3a2a2fb4a0a6a..25bdc880ba712 100644 --- a/task-sdk/src/airflow/sdk/__init__.py +++ b/task-sdk/src/airflow/sdk/__init__.py @@ -56,6 +56,7 @@ "DeltaDataIntervalTimetable", "DeltaTriggerTimetable", "EdgeModifier", + "emit_dag_warning", "EventsTimetable", "ExceptionRetryPolicy", "FanOutMapper", @@ -63,7 +64,6 @@ "HourWindow", "IdentityMapper", "Label", - "LoopDecision", "Metadata", "MinimumCount", "MonthWindow", @@ -166,7 +166,6 @@ from airflow.sdk.definitions.decorators import result, setup, task, teardown from airflow.sdk.definitions.decorators.task_group import task_group from airflow.sdk.definitions.edges import EdgeModifier, Label - from airflow.sdk.definitions.loop import LoopDecision from airflow.sdk.definitions.param import Param, ParamsDict from airflow.sdk.definitions.partition_mappers.allowed_key import AllowedKeyMapper from airflow.sdk.definitions.partition_mappers.base import ( @@ -274,6 +273,7 @@ "DeltaDataIntervalTimetable": ".definitions.timetables.interval", "DeltaTriggerTimetable": ".definitions.timetables.trigger", "EdgeModifier": ".definitions.edges", + "emit_dag_warning": ".definitions.parse_warning", "EventsTimetable": ".definitions.timetables.events", "ExceptionRetryPolicy": ".definitions.retry_policy", "FanOutMapper": ".definitions.partition_mappers.temporal", @@ -281,7 +281,6 @@ "HourWindow": ".definitions.partition_mappers.window", "IdentityMapper": ".definitions.partition_mappers.identity", "Label": ".definitions.edges", - "LoopDecision": ".definitions.loop", "Metadata": ".definitions.asset.metadata", "MinimumCount": ".definitions.partition_mappers.wait_policy", "MonthWindow": ".definitions.partition_mappers.window", diff --git a/task-sdk/src/airflow/sdk/definitions/_internal/contextmanager.py b/task-sdk/src/airflow/sdk/definitions/_internal/contextmanager.py index f0fbdc306e6f3..05b3ff5374333 100644 --- a/task-sdk/src/airflow/sdk/definitions/_internal/contextmanager.py +++ b/task-sdk/src/airflow/sdk/definitions/_internal/contextmanager.py @@ -23,6 +23,7 @@ from typing import TYPE_CHECKING, Any, Generic, TypeVar from airflow.sdk.definitions.dag import DAG +from airflow.sdk.definitions.parse_warning_record import DagParseWarningRecord from airflow.sdk.definitions.taskgroup import TaskGroup if TYPE_CHECKING: @@ -110,6 +111,7 @@ class DagContext(ContextStack[DAG]): autoregistered_dags: set[tuple[DAG, ModuleType]] = set() current_autoregister_module_name: str | None = None + parse_warnings: list[DagParseWarningRecord] = [] @classmethod def pop(cls) -> DAG | None: diff --git a/task-sdk/src/airflow/sdk/definitions/parse_warning.py b/task-sdk/src/airflow/sdk/definitions/parse_warning.py new file mode 100644 index 0000000000000..0fd2bf83d9d65 --- /dev/null +++ b/task-sdk/src/airflow/sdk/definitions/parse_warning.py @@ -0,0 +1,55 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import re + +from airflow.sdk.definitions.parse_warning_record import DagParseWarningRecord + +_PARSE_WARNING_TYPE_RE = re.compile(r"^[a-z][a-z0-9_]*:[a-z0-9_.\-]+$", re.ASCII) +_DEFAULT_WARNING_TYPE = "user:dynamic_dag_failure" + + +def emit_dag_warning( + dag_id: str, + message: str, + *, + warning_type: str = _DEFAULT_WARNING_TYPE, + stacktrace: str | None = None, +) -> None: + """Record a parse-time warning for a Dag id that could not be loaded (#29275).""" + if not dag_id: + raise ValueError("dag_id must be non-empty") + from airflow.sdk.definitions._internal.contextmanager import DagContext + + if DagContext.current_autoregister_module_name is None: + raise RuntimeError("emit_dag_warning() can only be called while a Dag file is being parsed") + if not _PARSE_WARNING_TYPE_RE.fullmatch(warning_type) or len(warning_type) > 50: + raise ValueError(f"invalid warning_type: {warning_type!r}") + if stacktrace: + message = f"{message}\n\n{stacktrace}" + DagContext.parse_warnings.append( + DagParseWarningRecord(dag_id=dag_id, message=message, warning_type=warning_type) + ) + + +def drain_parse_warnings() -> list[DagParseWarningRecord]: + from airflow.sdk.definitions._internal.contextmanager import DagContext + + warnings = DagContext.parse_warnings + DagContext.parse_warnings = [] + return warnings diff --git a/task-sdk/src/airflow/sdk/definitions/parse_warning_record.py b/task-sdk/src/airflow/sdk/definitions/parse_warning_record.py new file mode 100644 index 0000000000000..20deb144ccce0 --- /dev/null +++ b/task-sdk/src/airflow/sdk/definitions/parse_warning_record.py @@ -0,0 +1,10 @@ +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass +class DagParseWarningRecord: + dag_id: str + message: str + warning_type: str = "user:dynamic_dag_failure" diff --git a/task-sdk/src/airflow/sdk/importers/base.py b/task-sdk/src/airflow/sdk/importers/base.py index 390c28bfc2fe5..4940b9eb8c55a 100644 --- a/task-sdk/src/airflow/sdk/importers/base.py +++ b/task-sdk/src/airflow/sdk/importers/base.py @@ -29,6 +29,7 @@ from airflow.sdk._shared.module_loading.file_discovery import find_path_from_directory from airflow.sdk.configuration import conf +from airflow.sdk.definitions.parse_warning_record import DagParseWarningRecord from airflow.sdk.exceptions import AirflowConfigException from airflow.sdk.execution_time.coordinator import get_coordinator_manager @@ -220,6 +221,7 @@ class DagImportResult: errors: list[DagImportError] = field(default_factory=list) skipped_definitions: list[DagDefinition] = field(default_factory=list) warnings: list[DagImportWarning] = field(default_factory=list) + parse_warnings: list[DagParseWarningRecord] = field(default_factory=list) dependencies: list[DagDefinition] = field(default_factory=list) @property diff --git a/task-sdk/src/airflow/sdk/importers/python_importer.py b/task-sdk/src/airflow/sdk/importers/python_importer.py index 502745792f24e..fa1055e3c693d 100644 --- a/task-sdk/src/airflow/sdk/importers/python_importer.py +++ b/task-sdk/src/airflow/sdk/importers/python_importer.py @@ -178,6 +178,7 @@ def import_definition( """Import DAGs from a Python DAG definition.""" result = DagImportResult(definition=definition) DagContext.autoregistered_dags.clear() + DagContext.parse_warnings.clear() captured_warnings: list[warnings.WarningMessage] = [] try: @@ -211,6 +212,9 @@ def import_definition( ) self._process_modules(modules, result, bundle=bundle) + from airflow.sdk.definitions.parse_warning import drain_parse_warnings + + result.parse_warnings.extend(drain_parse_warnings()) return result def get_source_code(self, definition: DagDefinition, dag_id: str | None = None) -> DagSourceCode: @@ -250,9 +254,10 @@ def _handle_sigsegv(signum, frame): if mod_name in sys.modules: del sys.modules[mod_name] + DagContext.current_autoregister_module_name = mod_name + def parse() -> list[types.ModuleType]: try: - DagContext.current_autoregister_module_name = mod_name with definition.import_context(): loader: importlib.abc.Loader if get_file_suffix(definition) == ".pyc": @@ -270,6 +275,7 @@ def parse() -> list[types.ModuleType]: except BaseException as e: sys.modules.pop(mod_name, None) DagContext.autoregistered_dags.clear() + DagContext.parse_warnings.clear() log.exception("Failed to import: %r", definition) if self._dagbag_import_error_tracebacks: stacktrace = traceback.format_exc(limit=-self._dagbag_import_error_traceback_depth) @@ -284,10 +290,6 @@ def parse() -> list[types.ModuleType]: ) ) return [] - finally: - # Reset on every exit: DagContext.pop() looks this name up in sys.modules, which may not - # hold it if the load did not complete. - DagContext.current_autoregister_module_name = None dagbag_import_timeout: float try: @@ -326,6 +328,7 @@ def _process_modules( } top_level_dags.update(DagContext.autoregistered_dags) + DagContext.current_autoregister_module_name = None DagContext.autoregistered_dags.clear() for dag, _mod in top_level_dags: From 71302a7491c94af76a0c37d00c263a358ef0e489 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Mon, 26 Oct 2026 10:03:55 +0530 Subject: [PATCH 02/21] ui change --- .../core_api/datamodels/dag_warning.py | 28 ++++++------ .../core_api/routes/public/dag_warning.py | 6 ++- .../airflow/api_fastapi/core_api/security.py | 4 +- .../ui/public/i18n/locales/en/components.json | 3 +- .../ui/src/components/DAGWarningsModal.tsx | 16 +++---- .../ui/src/components/WarningAlert.tsx | 19 ++++---- .../Dashboard/Stats/ParseFailureWarnings.tsx | 43 ++++++++++--------- .../example_dags/example_repro_29275.py | 2 +- 8 files changed, 65 insertions(+), 56 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_warning.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_warning.py index e49125dd87b99..fbb5a939d7916 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_warning.py +++ b/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, model_validator - 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,16 +31,20 @@ class DAGWarningResponse(BaseModel): warning_type: DagWarningTypeValue message: str timestamp: datetime - dag_display_name: str | None = Field( - default=None, validation_alias=AliasPath("dag_model", "dag_display_name") - ) - - -@model_validator(mode="after") -def default_dag_display_name(self) -> DAGWarningResponse: - if not self.dag_display_name: - self.dag_display_name = self.dag_id - return self + 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): diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_warning.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_warning.py index 0096b82d1ba3e..66a3c86289724 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_warning.py +++ b/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, ) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/security.py b/airflow-core/src/airflow/api_fastapi/core_api/security.py index edb4f289291a7..0b91ea1d701f3 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/security.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/security.py @@ -315,14 +315,14 @@ 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: + 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 select.where( + return statement.where( or_( DagWarning.dag_id.in_(authorized), and_( diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/components.json b/airflow-core/src/airflow/ui/public/i18n/locales/en/components.json index 40d5f6ffff6b1..94693c1acbd09 100644 --- a/airflow-core/src/airflow/ui/public/i18n/locales/en/components.json +++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/components.json @@ -63,7 +63,8 @@ "duplicate dag id": "Duplicate Dag ID", "non-existent pool": "Non-existent pool", "runtime varying value": "Runtime-varying value", - "task group cycle": "Task group cycle" + "task group cycle": "Task group cycle", + "user:dynamic_dag_failure": "Dag failed to load" }, "warning_one": "1 Warning", "warning_other": "{{count, number}} Warnings" diff --git a/airflow-core/src/airflow/ui/src/components/DAGWarningsModal.tsx b/airflow-core/src/airflow/ui/src/components/DAGWarningsModal.tsx index e246137335071..18a55da583e2d 100644 --- a/airflow-core/src/airflow/ui/src/components/DAGWarningsModal.tsx +++ b/airflow-core/src/airflow/ui/src/components/DAGWarningsModal.tsx @@ -65,12 +65,13 @@ export const DAGWarningsModal = ({ data-testid="dag-warnings-modal" headerProps={{ children: ( - + - {heading} + {heading} ), }} + bodyProps={{ gap:4 }} onOpenChange={onClose} open={open} scrollBehavior="inside" @@ -97,12 +98,11 @@ export const DAGWarningsModal = ({ )} - {warnings?.map((warning) => ( - - - - - ))} + + {warnings?.map((warning) => ( + + ))} + ); }; diff --git a/airflow-core/src/airflow/ui/src/components/WarningAlert.tsx b/airflow-core/src/airflow/ui/src/components/WarningAlert.tsx index cb5cfaa08fa82..f490d1afe656b 100644 --- a/airflow-core/src/airflow/ui/src/components/WarningAlert.tsx +++ b/airflow-core/src/airflow/ui/src/components/WarningAlert.tsx @@ -22,6 +22,9 @@ import type { DAGWarningResponse } from "openapi/requests/types.gen"; import { StatusMessage } from "src/components/StatusMessage"; + +const PARSE_FAILURE_WARNING_TYPE = "user:dynamic_dag_failure"; + type Props = { readonly warning?: DAGWarningResponse; }; @@ -35,13 +38,13 @@ export const WarningAlert = ({ warning }: Props) => { // The warning's type is its short form, but it arrives as the backend's own lowercase enum value, // which has no business being a heading in any language. An unmapped one falls back to the word. - return ( - + const heading = + warning.warning_type === PARSE_FAILURE_WARNING_TYPE + ? (warning.dag_display_name ?? warning.dag_id) + : translate(`dagWarnings.types.${warning.warning_type}`, { + defaultValue: translate("dagWarnings.warning_one") + } ); + return ; + }; diff --git a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx index b5dd815176844..5294c059e9db8 100644 --- a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx +++ b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx @@ -1,4 +1,5 @@ -/*! +/* +! * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information @@ -16,53 +17,53 @@ * specific language governing permissions and limitations * under the License. */ -import { useDisclosure } from "@chakra-ui/react"; +import { Skeleton, useDisclosure } from "@chakra-ui/react"; import { useTranslation } from "react-i18next"; import { LuFileWarning } from "react-icons/lu"; - import { useDagWarningServiceListDagWarnings } from "openapi/queries"; - import { DAGWarningsModal } from "src/components/DAGWarningsModal"; +import type { ExpandedApiError } from "src/components/ErrorAlert"; import { StatsCard } from "src/components/StatsCard"; - import { useAutoRefresh } from "src/utils"; - const PARSE_FAILURE_WARNING_TYPE = "user:dynamic_dag_failure"; - export const ParseFailureWarnings = () => { const { onClose, onOpen, open } = useDisclosure(); const refetchInterval = useAutoRefresh({ checkPendingRuns: true }); - const { i18n, t: translate } = useTranslation("dashboard"); - const { data, isLoading } = useDagWarningServiceListDagWarnings( + const { i18n, t: translate } = useTranslation(["dashboard", "components"]); + const { data, error, isLoading } = useDagWarningServiceListDagWarnings( { limit: 100, warningType: PARSE_FAILURE_WARNING_TYPE }, undefined, { refetchInterval }, ); const count = data?.total_entries ?? 0; - - if (!isLoading && count === 0) { + if (isLoading) { + return ; + } + if (Boolean(error) && (error as ExpandedApiError).status === 403) { + return undefined; + } + if (count === 0) { return undefined; } - return ( <> - } - isLoading={isLoading} + isLoading={false} isRTL={i18n.dir() === "rtl"} - label={translate("stats.parseFailureWarnings", { count })} + label={translate("dashboard:stats.parseFailureWarnings", { count })} onClick={onOpen} /> - ({ - ...warning, - message: `${warning.dag_id}: ${warning.message}`, - }))} + warnings={data?.dag_warnings} /> ); -}; +}; \ No newline at end of file diff --git a/providers/standard/src/airflow/providers/standard/example_dags/example_repro_29275.py b/providers/standard/src/airflow/providers/standard/example_dags/example_repro_29275.py index f199cfc693047..2e73c26feb5bc 100644 --- a/providers/standard/src/airflow/providers/standard/example_dags/example_repro_29275.py +++ b/providers/standard/src/airflow/providers/standard/example_dags/example_repro_29275.py @@ -25,5 +25,5 @@ def t(): dynamic_dag() except Exception: - emit_dag_warning(f"repro_{dag_id}", f"Failed to create dynamic DAG {dag_id}", stacktrace=None) + emit_dag_warning(f"repro_{dag_id}", f"Failed to create dynamic DAG {dag_id}") log.exception("Failed to create dynamic DAG %s", dag_id) From 6687e422801a46c6226aae18bd79a20884f91158 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Mon, 26 Oct 2026 11:02:56 +0530 Subject: [PATCH 03/21] remove unwanted files --- .../example_ext_task_sensor_child.py | 46 ------------------- .../example_ext_task_sensor_parent.py | 33 ------------- .../ui/src/components/DAGWarningsModal.tsx | 4 +- .../Dashboard/Stats/ParseFailureWarnings.tsx | 11 ++--- .../example_dags/example_repro_29275.py | 3 +- 5 files changed, 8 insertions(+), 89 deletions(-) delete mode 100644 airflow-core/src/airflow/example_dags/example_ext_task_sensor_child.py delete mode 100644 airflow-core/src/airflow/example_dags/example_ext_task_sensor_parent.py diff --git a/airflow-core/src/airflow/example_dags/example_ext_task_sensor_child.py b/airflow-core/src/airflow/example_dags/example_ext_task_sensor_child.py deleted file mode 100644 index 1d12096bc2cdd..0000000000000 --- a/airflow-core/src/airflow/example_dags/example_ext_task_sensor_child.py +++ /dev/null @@ -1,46 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. - -from __future__ import annotations - -from datetime import datetime, timedelta - -from airflow.sdk import DAG -from airflow.sensors.external_task import ExternalTaskSensor - - -def get_previous_execution_date(logical_date, **kwargs): - """Get the previous execution date of the parent DAG.""" - return logical_date - timedelta(minutes=5) - - -with DAG( - dag_id="example_ext_task_sensor_child", - start_date=datetime(2026, 2, 11), - schedule="42 00 * * *", - catchup=False, - tags=["example", "ext-sensor"], - description="This DAG is used to demonstrate the ExternalTaskSensor. It is used in conjunction with example_ext_task_sensor_parent.py", -) as dag: - wait_for_parent = ExternalTaskSensor( - task_id="wait_for_parent", - external_dag_id="example_ext_task_sensor_parent", - execution_date_fn=get_previous_execution_date, - poke_interval=30, - mode="poke", - timeout=300, - ) diff --git a/airflow-core/src/airflow/example_dags/example_ext_task_sensor_parent.py b/airflow-core/src/airflow/example_dags/example_ext_task_sensor_parent.py deleted file mode 100644 index c8dd56822c573..0000000000000 --- a/airflow-core/src/airflow/example_dags/example_ext_task_sensor_parent.py +++ /dev/null @@ -1,33 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. - -from __future__ import annotations - -from datetime import datetime - -from airflow.operators.empty import EmptyOperator -from airflow.sdk import DAG - -with DAG( - dag_id="example_ext_task_sensor_parent", - start_date=datetime(2026, 2, 11), - schedule="37 00 * * *", - catchup=False, - tags=["example", "ext-sensor"], - description="This DAG is used to demonstrate the ExternalTaskSensor. It is used in conjunction with example_ext_task_sensor_child.py", -) as dag: - EmptyOperator(task_id="parent_task") diff --git a/airflow-core/src/airflow/ui/src/components/DAGWarningsModal.tsx b/airflow-core/src/airflow/ui/src/components/DAGWarningsModal.tsx index 18a55da583e2d..e738d5dd81837 100644 --- a/airflow-core/src/airflow/ui/src/components/DAGWarningsModal.tsx +++ b/airflow-core/src/airflow/ui/src/components/DAGWarningsModal.tsx @@ -71,7 +71,7 @@ export const DAGWarningsModal = ({ ), }} - bodyProps={{ gap:4 }} + bodyProps={{ gap: 4 }} onOpenChange={onClose} open={open} scrollBehavior="inside" @@ -100,7 +100,7 @@ export const DAGWarningsModal = ({ )} {warnings?.map((warning) => ( - + ))} diff --git a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx index 5294c059e9db8..a9bb265c80d8c 100644 --- a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx +++ b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx @@ -1,5 +1,4 @@ -/* -! +/*! * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information @@ -29,7 +28,7 @@ const PARSE_FAILURE_WARNING_TYPE = "user:dynamic_dag_failure"; export const ParseFailureWarnings = () => { const { onClose, onOpen, open } = useDisclosure(); const refetchInterval = useAutoRefresh({ checkPendingRuns: true }); - const { i18n, t: translate } = useTranslation(["dashboard", "components"]); + const { i18n, t: translate } = useTranslation(["dashboard"]); const { data, error, isLoading } = useDagWarningServiceListDagWarnings( { limit: 100, warningType: PARSE_FAILURE_WARNING_TYPE }, undefined, @@ -47,8 +46,7 @@ export const ParseFailureWarnings = () => { } return ( <> - < -StatsCard + } @@ -57,8 +55,7 @@ StatsCard label={translate("dashboard:stats.parseFailureWarnings", { count })} onClick={onOpen} /> - < -DAGWarningsModal + Date: Mon, 26 Oct 2026 12:15:18 +0530 Subject: [PATCH 04/21] formatting --- .../core_api/openapi/v2-rest-api-generated.yaml | 5 ++--- .../airflow/ui/openapi-gen/requests/schemas.gen.ts | 11 +++++++++-- .../src/airflow/ui/openapi-gen/requests/types.gen.ts | 2 +- 3 files changed, 12 insertions(+), 6 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml index 1f513ae15e8f7..56cd607999978 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml @@ -15553,9 +15553,7 @@ components: format: date-time title: Timestamp dag_display_name: - anyOf: - - type: string - - type: 'null' + type: string title: Dag Display Name type: object required: @@ -15563,6 +15561,7 @@ components: - warning_type - message - timestamp + - dag_display_name title: DAGWarningResponse description: Dag Warning serializer for responses. DagBundleCollectionResponse: diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts index d121d52083678..7c072c0515e72 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts @@ -4728,12 +4728,19 @@ export const $DAGWarningResponse = { title: 'Timestamp' }, dag_display_name: { - type: 'string', + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], title: 'Dag Display Name' } }, type: 'object', - required: ['dag_id', 'warning_type', 'message', 'timestamp', 'dag_display_name'], + required: ['dag_id', 'warning_type', 'message', 'timestamp'], title: 'DAGWarningResponse', description: 'Dag Warning serializer for responses.' } as const; diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts index 0a90bf284f5d4..8412d3f54971d 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts @@ -1222,7 +1222,7 @@ export type DAGWarningResponse = { warning_type: DagWarningType | string; message: string; timestamp: string; - dag_display_name: string; + dag_display_name?: string | null; }; /** From 5fa8c1c10652eabc51ab873cdb3ca73a6958e67f Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Mon, 26 Oct 2026 12:35:47 +0530 Subject: [PATCH 05/21] remove unwanted files --- .../example_dags/example_repro_29275.py | 30 ------------------- 1 file changed, 30 deletions(-) delete mode 100644 providers/standard/src/airflow/providers/standard/example_dags/example_repro_29275.py diff --git a/providers/standard/src/airflow/providers/standard/example_dags/example_repro_29275.py b/providers/standard/src/airflow/providers/standard/example_dags/example_repro_29275.py deleted file mode 100644 index 8952656009915..0000000000000 --- a/providers/standard/src/airflow/providers/standard/example_dags/example_repro_29275.py +++ /dev/null @@ -1,30 +0,0 @@ -from __future__ import annotations - -import logging -from datetime import datetime -import traceback - - -from airflow.sdk import dag, task, 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) From b32732c127ed9bb46e5403cf5748385fa1af902c Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Mon, 26 Oct 2026 12:52:19 +0530 Subject: [PATCH 06/21] generated files --- .../airflow/ui/openapi-gen/requests/schemas.gen.ts | 11 ++--------- .../src/airflow/ui/openapi-gen/requests/types.gen.ts | 2 +- 2 files changed, 3 insertions(+), 10 deletions(-) diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts index 7c072c0515e72..d121d52083678 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts @@ -4728,19 +4728,12 @@ export const $DAGWarningResponse = { title: 'Timestamp' }, dag_display_name: { - anyOf: [ - { - type: 'string' - }, - { - type: 'null' - } - ], + type: 'string', title: 'Dag Display Name' } }, type: 'object', - required: ['dag_id', 'warning_type', 'message', 'timestamp'], + required: ['dag_id', 'warning_type', 'message', 'timestamp', 'dag_display_name'], title: 'DAGWarningResponse', description: 'Dag Warning serializer for responses.' } as const; diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts index 8412d3f54971d..0a90bf284f5d4 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts @@ -1222,7 +1222,7 @@ export type DAGWarningResponse = { warning_type: DagWarningType | string; message: string; timestamp: string; - dag_display_name?: string | null; + dag_display_name: string; }; /** From 8ea56dee3f776b597e5230ca4c8a8af3e2d3ccb1 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Mon, 26 Oct 2026 14:58:03 +0530 Subject: [PATCH 07/21] fix pre commit checks --- .../src/airflow/dag_processing/lang_sdk_processor.py | 4 ++-- airflow-core/src/airflow/dag_processing/processor.py | 11 ++++++----- go-sdk/pkg/execution/genmodels/models.gen.go | 5 +++++ go-sdk/schema/supervisor-schema.json | 8 ++++++++ .../src/airflow/sdk/execution_time/schema/schema.json | 8 ++++++++ .../sdk/execution_time/schema/versions/__init__.py | 2 ++ .../sdk/execution_time/schema/versions/v2026_10_30.py | 7 +++++++ ts-sdk/schema/supervisor-schema.json | 8 ++++++++ ts-sdk/src/generated/supervisor.ts | 4 ++++ 9 files changed, 50 insertions(+), 7 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/lang_sdk_processor.py b/airflow-core/src/airflow/dag_processing/lang_sdk_processor.py index 41ecf8b66a65b..9e753e61059c1 100644 --- a/airflow-core/src/airflow/dag_processing/lang_sdk_processor.py +++ b/airflow-core/src/airflow/dag_processing/lang_sdk_processor.py @@ -42,7 +42,7 @@ BaseDagFileProcessorProcess, DagFileParseRequest, DagFileParsingResult, - ToManager, + ToManagerMessage, ) from airflow.exceptions import DeserializationError from airflow.models.pool import Pool @@ -101,7 +101,7 @@ class LangSDKRuntimeSchemaVersion(BaseModel): type: Literal["LangSDKRuntimeSchemaVersion"] = "LangSDKRuntimeSchemaVersion" -LangSDKToManager = Annotated[LangSDKRuntimeSchemaVersion | ToManager, Field(discriminator="type")] +LangSDKToManager = Annotated[LangSDKRuntimeSchemaVersion | ToManagerMessage, Field(discriminator="type")] def _get_import_timeout(path: str) -> float | None: diff --git a/airflow-core/src/airflow/dag_processing/processor.py b/airflow-core/src/airflow/dag_processing/processor.py index 76221cd88908f..e2e19af72f0e9 100644 --- a/airflow-core/src/airflow/dag_processing/processor.py +++ b/airflow-core/src/airflow/dag_processing/processor.py @@ -136,7 +136,7 @@ class DagFileParsingResult(BaseModel): type: Literal["DagFileParsingResult"] = "DagFileParsingResult" -ToManager = Annotated[ +ToManagerMessage = ( DagFileParsingResult | GetConnection | GetVariable @@ -152,9 +152,10 @@ class DagFileParsingResult(BaseModel): | GetXComCount | GetXComSequenceItem | GetXComSequenceSlice - | MaskSecret, - Field(discriminator="type"), -] + | MaskSecret +) + +ToManager = Annotated[ToManagerMessage, Field(discriminator="type")] ToDagProcessor = Annotated[ DagFileParseRequest @@ -599,7 +600,6 @@ class BaseDagFileProcessorProcess(WatchedSubprocess, LoggingMixin): ) """The file's parse log, which the process closes. Without one, the output is discarded.""" parsing_result: DagFileParsingResult | None = None - decoder: ClassVar[TypeAdapter[ToManager]] = TypeAdapter[ToManager](ToManager) had_callbacks: bool = False # Track if this process was started with callbacks to prevent stale DAG detection false positives client: Client | None = None @@ -724,6 +724,7 @@ class DagFileProcessorProcess(BaseDagFileProcessorProcess): """ client: Client + decoder: ClassVar[TypeAdapter[ToManager]] = TypeAdapter[ToManager](ToManager) _request_handlers: ClassVar[dict[type[BaseModel], RequestHandler[DagFileProcessorProcess]]] = { **BaseDagFileProcessorProcess._common_request_handlers, diff --git a/go-sdk/pkg/execution/genmodels/models.gen.go b/go-sdk/pkg/execution/genmodels/models.gen.go index f95ceb53d629a..3424f58f83be1 100644 --- a/go-sdk/pkg/execution/genmodels/models.gen.go +++ b/go-sdk/pkg/execution/genmodels/models.gen.go @@ -416,6 +416,9 @@ type DagFileParsingResult struct { // ImportErrors corresponds to the JSON schema field "import_errors". ImportErrors *ImportErrors `msgpack:"import_errors,omitempty"` + // ParseWarnings corresponds to the JSON schema field "parse_warnings". + ParseWarnings []DagFileParsingResultParseWarningsElem `msgpack:"parse_warnings,omitempty"` + // ParsedDefinitions corresponds to the JSON schema field "parsed_definitions". ParsedDefinitions []string `msgpack:"parsed_definitions,omitempty"` @@ -429,6 +432,8 @@ type DagFileParsingResult struct { Warnings *Warnings `msgpack:"warnings,omitempty"` } +type DagFileParsingResultParseWarningsElem map[string]interface{} + type DagResult struct { // BundleName corresponds to the JSON schema field "bundle_name". BundleName interface{} `msgpack:"bundle_name"` diff --git a/go-sdk/schema/supervisor-schema.json b/go-sdk/schema/supervisor-schema.json index ebe8c2ab65ce8..419e7301800fc 100644 --- a/go-sdk/schema/supervisor-schema.json +++ b/go-sdk/schema/supervisor-schema.json @@ -1022,6 +1022,14 @@ "title": "Parsed Definitions", "type": "array" }, + "parse_warnings": { + "items": { + "additionalProperties": true, + "type": "object" + }, + "title": "Parse Warnings", + "type": "array" + }, "dag_source_codes": { "additionalProperties": { "$ref": "#/$defs/DagSourceCode" diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json index ebe8c2ab65ce8..419e7301800fc 100644 --- a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json +++ b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json @@ -1022,6 +1022,14 @@ "title": "Parsed Definitions", "type": "array" }, + "parse_warnings": { + "items": { + "additionalProperties": true, + "type": "object" + }, + "title": "Parse Warnings", + "type": "array" + }, "dag_source_codes": { "additionalProperties": { "$ref": "#/$defs/DagSourceCode" diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py b/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py index 27ff074bb237b..33386e3c3039e 100644 --- a/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py +++ b/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py @@ -44,6 +44,7 @@ def get_bundle() -> VersionBundle: AddExpandedTICountToSupervisorTIRunContext, AddFirstTaskRescheduleStartDateToSupervisorTIRunContext, AddLoopRuntime, + AddParseWarningsToDagFileParsingResult, AddRegionCoordinatesToSupervisorTaskInstance, AddRetryReasonToTaskState, ) @@ -56,6 +57,7 @@ def get_bundle() -> VersionBundle: AddLoopRuntime, AddRetryReasonToTaskState, AddDagDefinitionsToDagFileParsingResult, + AddParseWarningsToDagFileParsingResult, AddRegionCoordinatesToSupervisorTaskInstance, AddFirstTaskRescheduleStartDateToSupervisorTIRunContext, AddExpandedTICountToSupervisorTIRunContext, diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py index ddc99b274b987..aaa0b2d8cffb8 100644 --- a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py +++ b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py @@ -105,6 +105,13 @@ class AddDagDefinitionsToDagFileParsingResult(VersionChange): ) +class AddParseWarningsToDagFileParsingResult(VersionChange): + description = __doc__ + instructions_to_migrate_to_previous_version = ( + schema(DagFileParsingResult).field("parse_warnings").didnt_exist, + ) + + class AddFirstTaskRescheduleStartDateToSupervisorTIRunContext(VersionChange): """ Add the ``first_task_reschedule_start_date`` field carrying the earliest reschedule start date. diff --git a/ts-sdk/schema/supervisor-schema.json b/ts-sdk/schema/supervisor-schema.json index ebe8c2ab65ce8..419e7301800fc 100644 --- a/ts-sdk/schema/supervisor-schema.json +++ b/ts-sdk/schema/supervisor-schema.json @@ -1022,6 +1022,14 @@ "title": "Parsed Definitions", "type": "array" }, + "parse_warnings": { + "items": { + "additionalProperties": true, + "type": "object" + }, + "title": "Parse Warnings", + "type": "array" + }, "dag_source_codes": { "additionalProperties": { "$ref": "#/$defs/DagSourceCode" diff --git a/ts-sdk/src/generated/supervisor.ts b/ts-sdk/src/generated/supervisor.ts index 69285544734c5..5af9bc86cea6a 100644 --- a/ts-sdk/src/generated/supervisor.ts +++ b/ts-sdk/src/generated/supervisor.ts @@ -292,6 +292,9 @@ export type ImportErrors = { [k: string]: string; } | null; export type ParsedDefinitions = string[]; +export type ParseWarnings = { + [k: string]: unknown; +}[]; export type SourceCode = string; export type Language = string; export type Type16 = "DagFileParsingResult"; @@ -1162,6 +1165,7 @@ export interface DagFileParsingResult { warnings?: Warnings; import_errors?: ImportErrors; parsed_definitions?: ParsedDefinitions; + parse_warnings?: ParseWarnings; dag_source_codes?: DagSourceCodes; type?: Type16; } From 09abe706b79d1b54c01d5a2544f47bb9cfeedee9 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Mon, 26 Oct 2026 17:34:20 +0530 Subject: [PATCH 08/21] revert --- .../src/airflow/dag_processing/collection.py | 27 +++++++++++++++++++ .../ui/public/i18n/locales/en/dashboard.json | 6 ++--- .../ui/src/components/DAGWarningsModal.tsx | 4 +-- .../ui/src/components/WarningAlert.tsx | 16 +++++------ .../Dashboard/Stats/ParseFailureWarnings.tsx | 9 ++++++- airflow-core/src/airflow/utils/db.py | 2 +- .../tests/unit/models/test_dagwarning.py | 2 +- .../sdk/definitions/parse_warning_record.py | 19 +++++++++++++ 8 files changed, 68 insertions(+), 17 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/collection.py b/airflow-core/src/airflow/dag_processing/collection.py index 9ed35f252be82..580d2b879f35d 100644 --- a/airflow-core/src/airflow/dag_processing/collection.py +++ b/airflow-core/src/airflow/dag_processing/collection.py @@ -607,6 +607,32 @@ def _reject_other_teams_plugin_classes( return accepted +def _reject_dags_with_unloadable_timetables( + bundle_name: str, + dags: Collection[LazyDeserializedDAG], + import_errors: dict[tuple[str, str], str], +) -> list[LazyDeserializedDAG]: + """ + Drop Dags whose timetable raises when rebuilt, recording an import error for each. + + A custom timetable runs user code when it is deserialized, for example ``Variable.get`` in + ``__init__``. Checking here keeps one failing Dag from aborting the save of every Dag in its file. + """ + accepted: list[LazyDeserializedDAG] = [] + for dag in dags: + try: + dag.timetable + except Exception: + log.exception("Failed to load timetable of Dag", dag_id=dag.dag_id) + depth = conf.getint("core", "dagbag_import_error_traceback_depth") + import_errors[(bundle_name, dag.relative_fileloc)] = ( + f"Failed to load the timetable of Dag '{dag.dag_id}':\n{traceback.format_exc(limit=-depth)}" + ) + continue + accepted.append(dag) + return accepted + + def update_dag_parsing_results_in_db( bundle_name: str, bundle_version: str | None, @@ -651,6 +677,7 @@ def update_dag_parsing_results_in_db( without an entry have their source read from ``fileloc``. """ accepted = _reject_other_teams_plugin_classes(bundle_name, dags, import_errors, session=session) + accepted = _reject_dags_with_unloadable_timetables(bundle_name, accepted, import_errors) if len(accepted) != len(dags): # A rejected Dag may have no ``dag`` row yet, and dag_warning has a foreign key to it. rejected_ids = {dag.dag_id for dag in dags} - {dag.dag_id for dag in accepted} diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/dashboard.json b/airflow-core/src/airflow/ui/public/i18n/locales/en/dashboard.json index 2fa134fef9bcc..779fa10c4030b 100644 --- a/airflow-core/src/airflow/ui/public/i18n/locales/en/dashboard.json +++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/dashboard.json @@ -71,13 +71,13 @@ "stats": { "activeDags": "Active Dags", "failedDags": "Failed Dags", + "parseFailureWarnings_one": "Dag load warning", + "parseFailureWarnings_other": "Dag load warnings", "queuedDags": "Queued Dags", "requiredActions": "Required Actions", "runningDags": "Running Dags", - "parseFailureWarnings_one": "Dag load warning", - "parseFailureWarnings_other": "Dag load warnings", "stats": "Stats" }, "uri": "Uri", "welcome": "Welcome" -} \ No newline at end of file +} diff --git a/airflow-core/src/airflow/ui/src/components/DAGWarningsModal.tsx b/airflow-core/src/airflow/ui/src/components/DAGWarningsModal.tsx index e738d5dd81837..7a3c639ff6f1a 100644 --- a/airflow-core/src/airflow/ui/src/components/DAGWarningsModal.tsx +++ b/airflow-core/src/airflow/ui/src/components/DAGWarningsModal.tsx @@ -62,6 +62,7 @@ export const DAGWarningsModal = ({ return ( ), }} - bodyProps={{ gap: 4 }} onOpenChange={onClose} open={open} scrollBehavior="inside" @@ -101,7 +101,7 @@ export const DAGWarningsModal = ({ {warnings?.map((warning) => ( - ))} + ))} ); diff --git a/airflow-core/src/airflow/ui/src/components/WarningAlert.tsx b/airflow-core/src/airflow/ui/src/components/WarningAlert.tsx index f490d1afe656b..f9872e2defef1 100644 --- a/airflow-core/src/airflow/ui/src/components/WarningAlert.tsx +++ b/airflow-core/src/airflow/ui/src/components/WarningAlert.tsx @@ -22,7 +22,6 @@ import type { DAGWarningResponse } from "openapi/requests/types.gen"; import { StatusMessage } from "src/components/StatusMessage"; - const PARSE_FAILURE_WARNING_TYPE = "user:dynamic_dag_failure"; type Props = { @@ -38,13 +37,12 @@ export const WarningAlert = ({ warning }: Props) => { // The warning's type is its short form, but it arrives as the backend's own lowercase enum value, // which has no business being a heading in any language. An unmapped one falls back to the word. - const heading = - warning.warning_type === PARSE_FAILURE_WARNING_TYPE - ? (warning.dag_display_name ?? warning.dag_id) - : translate(`dagWarnings.types.${warning.warning_type}`, { - defaultValue: translate("dagWarnings.warning_one") - } - ); - return ; + const heading = + warning.warning_type === PARSE_FAILURE_WARNING_TYPE + ? warning.dag_display_name + : translate(`dagWarnings.types.${warning.warning_type}`, { + defaultValue: translate("dagWarnings.warning_one"), + }); + return ; }; diff --git a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx index a9bb265c80d8c..1b0721206fec6 100644 --- a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx +++ b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx @@ -19,12 +19,17 @@ import { Skeleton, useDisclosure } from "@chakra-ui/react"; import { useTranslation } from "react-i18next"; import { LuFileWarning } from "react-icons/lu"; + import { useDagWarningServiceListDagWarnings } from "openapi/queries"; + import { DAGWarningsModal } from "src/components/DAGWarningsModal"; import type { ExpandedApiError } from "src/components/ErrorAlert"; import { StatsCard } from "src/components/StatsCard"; + import { useAutoRefresh } from "src/utils"; + const PARSE_FAILURE_WARNING_TYPE = "user:dynamic_dag_failure"; + export const ParseFailureWarnings = () => { const { onClose, onOpen, open } = useDisclosure(); const refetchInterval = useAutoRefresh({ checkPendingRuns: true }); @@ -35,6 +40,7 @@ export const ParseFailureWarnings = () => { { refetchInterval }, ); const count = data?.total_entries ?? 0; + if (isLoading) { return ; } @@ -44,6 +50,7 @@ export const ParseFailureWarnings = () => { if (count === 0) { return undefined; } + return ( <> { /> ); -}; \ No newline at end of file +}; diff --git a/airflow-core/src/airflow/utils/db.py b/airflow-core/src/airflow/utils/db.py index 266222bd070cf..46995030e5fa2 100644 --- a/airflow-core/src/airflow/utils/db.py +++ b/airflow-core/src/airflow/utils/db.py @@ -117,7 +117,7 @@ class MappedClassProtocol(Protocol): "3.1.8": "509b94a1042d", "3.2.0": "1d6611b6ab7c", "3.3.0": "d2f4e1b3c5a7", - "3.4.0": "ee86eed19e24", + "3.4.0": "b8f2c1d04e91", "3.5.0": "c71065a1b14f", } diff --git a/airflow-core/tests/unit/models/test_dagwarning.py b/airflow-core/tests/unit/models/test_dagwarning.py index c0675cdc68364..0c0acdce8a45b 100644 --- a/airflow-core/tests/unit/models/test_dagwarning.py +++ b/airflow-core/tests/unit/models/test_dagwarning.py @@ -24,8 +24,8 @@ from sqlalchemy import select from sqlalchemy.exc import OperationalError -from airflow.models import DagModel from airflow.dag_processing.collection import _update_file_scoped_parse_failure_warnings +from airflow.models import DagModel from airflow.models.dagwarning import ( PARSE_FAILURE_WARNING_TYPE, DagWarning, diff --git a/task-sdk/src/airflow/sdk/definitions/parse_warning_record.py b/task-sdk/src/airflow/sdk/definitions/parse_warning_record.py index 20deb144ccce0..34d947234af5f 100644 --- a/task-sdk/src/airflow/sdk/definitions/parse_warning_record.py +++ b/task-sdk/src/airflow/sdk/definitions/parse_warning_record.py @@ -1,3 +1,20 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + from __future__ import annotations from dataclasses import dataclass @@ -5,6 +22,8 @@ @dataclass class DagParseWarningRecord: + """A parse-time warning for a single DAG id emitted during Dag file import.""" + dag_id: str message: str warning_type: str = "user:dynamic_dag_failure" From 3ccca6cd2c05f526295040c25ed3d60e0cb13b68 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Mon, 26 Oct 2026 19:31:10 +0530 Subject: [PATCH 09/21] revert wanted changes --- airflow-core/src/airflow/dag_processing/manager.py | 2 +- airflow-core/src/airflow/dag_processing/processor.py | 9 ++++++++- task-sdk/src/airflow/sdk/__init__.py | 3 +++ .../src/airflow/sdk/definitions/parse_warning_record.py | 2 +- 4 files changed, 13 insertions(+), 3 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index 04cc765e90c7f..88a2e09ead013 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/airflow-core/src/airflow/dag_processing/manager.py @@ -1871,7 +1871,7 @@ def process_parse_results( normalized_bundle = normalize_name_for_stats(bundle_name) stats.timing( "dag_processing.last_duration", - stat.last_duration, + timedelta(seconds=stat.last_duration), tags=prune_dict( {"bundle_name": normalized_bundle, "file_name": file_name, "team_name": team_name} ), diff --git a/airflow-core/src/airflow/dag_processing/processor.py b/airflow-core/src/airflow/dag_processing/processor.py index e2e19af72f0e9..bf013b2734494 100644 --- a/airflow-core/src/airflow/dag_processing/processor.py +++ b/airflow-core/src/airflow/dag_processing/processor.py @@ -73,7 +73,11 @@ XComSequenceSliceResult, ) from airflow.sdk.execution_time.supervisor import WatchedSubprocess, register_request_method -from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance, _send_error_email_notification +from airflow.sdk.execution_time.task_runner import ( + RuntimeTaskInstance, + _send_error_email_notification, + check_xcom_backend_supports_loops, +) from airflow.sdk.importers import DagSourceCode # noqa: TC001 from airflow.serialization.serialized_objects import DagSerialization, LazyDeserializedDAG from airflow.utils.dag_version_inflation_checker import check_dag_file_stability @@ -296,6 +300,9 @@ def _serialize_dags( serialized_dags = [] for dag in bag.dags.values(): try: + # Only possible because importing the SDK resolves the XCom backend here; the worker check is the + # one that must stay, so drop this one rather than keep that import working for its sake. + check_xcom_backend_supports_loops(dag) data = DagSerialization.to_dict(dag) serialized_dags.append(LazyDeserializedDAG(data=data, last_loaded=dag.last_loaded)) except Exception: diff --git a/task-sdk/src/airflow/sdk/__init__.py b/task-sdk/src/airflow/sdk/__init__.py index 25bdc880ba712..b03f2ad88a09c 100644 --- a/task-sdk/src/airflow/sdk/__init__.py +++ b/task-sdk/src/airflow/sdk/__init__.py @@ -64,6 +64,7 @@ "HourWindow", "IdentityMapper", "Label", + "LoopDecision", "Metadata", "MinimumCount", "MonthWindow", @@ -166,6 +167,7 @@ from airflow.sdk.definitions.decorators import result, setup, task, teardown from airflow.sdk.definitions.decorators.task_group import task_group from airflow.sdk.definitions.edges import EdgeModifier, Label + from airflow.sdk.definitions.loop import LoopDecision from airflow.sdk.definitions.param import Param, ParamsDict from airflow.sdk.definitions.partition_mappers.allowed_key import AllowedKeyMapper from airflow.sdk.definitions.partition_mappers.base import ( @@ -281,6 +283,7 @@ "HourWindow": ".definitions.partition_mappers.window", "IdentityMapper": ".definitions.partition_mappers.identity", "Label": ".definitions.edges", + "LoopDecision": ".definitions.loop", "Metadata": ".definitions.asset.metadata", "MinimumCount": ".definitions.partition_mappers.wait_policy", "MonthWindow": ".definitions.partition_mappers.window", diff --git a/task-sdk/src/airflow/sdk/definitions/parse_warning_record.py b/task-sdk/src/airflow/sdk/definitions/parse_warning_record.py index 34d947234af5f..96f960c6be0db 100644 --- a/task-sdk/src/airflow/sdk/definitions/parse_warning_record.py +++ b/task-sdk/src/airflow/sdk/definitions/parse_warning_record.py @@ -22,7 +22,7 @@ @dataclass class DagParseWarningRecord: - """A parse-time warning for a single DAG id emitted during Dag file import.""" + """A parse-time warning for a single DAG id emitted during DAG file import.""" dag_id: str message: str From 073439cb294672ec20000391b8a4b3e8806d8808 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Mon, 26 Oct 2026 20:12:25 +0530 Subject: [PATCH 10/21] versions handling --- airflow-core/docs/migrations-ref.rst | 5 ++++- .../versions/0150_3_5_0_add_dag_parse_warning.py | 11 +++++++++-- airflow-core/src/airflow/utils/db.py | 4 ++-- 3 files changed, 15 insertions(+), 5 deletions(-) diff --git a/airflow-core/docs/migrations-ref.rst b/airflow-core/docs/migrations-ref.rst index 922105789ae4b..747e8c6065fe5 100644 --- a/airflow-core/docs/migrations-ref.rst +++ b/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 | +=========================+==================+===================+==============================================================+ -| ``c71065a1b14f`` (head) | ``ee86eed19e24`` | ``3.5.0`` | Add session_id and registration_id to Job. | +| ``b8f2c1d04e91`` (head) | ``c71065a1b14f`` | ``3.5.0`` | Extend dag_warning for parse-time failures (file scope, no | +| | | | dag FK). | ++-------------------------+------------------+-------------------+--------------------------------------------------------------+ +| ``c71065a1b14f`` | ``ee86eed19e24`` | ``3.5.0`` | Add session_id and registration_id to Job. | +-------------------------+------------------+-------------------+--------------------------------------------------------------+ | ``ee86eed19e24`` | ``9cffe90fa0c0`` | ``3.4.0`` | Add pending_partition_key to asset_partition_dag_run and | | | | | enforce single pending row per key. | diff --git a/airflow-core/src/airflow/migrations/versions/0150_3_5_0_add_dag_parse_warning.py b/airflow-core/src/airflow/migrations/versions/0150_3_5_0_add_dag_parse_warning.py index efd7a0e0ad0a0..fe60e918c5a97 100644 --- a/airflow-core/src/airflow/migrations/versions/0150_3_5_0_add_dag_parse_warning.py +++ b/airflow-core/src/airflow/migrations/versions/0150_3_5_0_add_dag_parse_warning.py @@ -15,7 +15,14 @@ # specific language governing permissions and limitations # under the License. -"""Extend dag_warning for parse-time failures (file scope, no dag FK).""" +""" +Extend dag_warning for parse-time failures (file scope, no dag FK). + +Revision ID: b8f2c1d04e91 +Revises: c71065a1b14f +Create Date: 2026-10-26 00:00:00.000000 + +""" from __future__ import annotations @@ -28,7 +35,7 @@ down_revision = "c71065a1b14f" branch_labels = None depends_on = None -airflow_version = "3.4.0" +airflow_version = "3.5.0" def upgrade(): diff --git a/airflow-core/src/airflow/utils/db.py b/airflow-core/src/airflow/utils/db.py index 46995030e5fa2..1f906d3b58604 100644 --- a/airflow-core/src/airflow/utils/db.py +++ b/airflow-core/src/airflow/utils/db.py @@ -117,8 +117,8 @@ class MappedClassProtocol(Protocol): "3.1.8": "509b94a1042d", "3.2.0": "1d6611b6ab7c", "3.3.0": "d2f4e1b3c5a7", - "3.4.0": "b8f2c1d04e91", - "3.5.0": "c71065a1b14f", + "3.4.0": "ee86eed19e24", + "3.5.0": "b8f2c1d04e91", } # Prefix used to identify tables holding data moved during migration. From dc035efa9ae45379a1fa6832efb3e64179ecbc80 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Wed, 28 Oct 2026 00:10:03 +0530 Subject: [PATCH 11/21] fix test failure --- task-sdk/src/airflow/sdk/importers/python_importer.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/task-sdk/src/airflow/sdk/importers/python_importer.py b/task-sdk/src/airflow/sdk/importers/python_importer.py index fa1055e3c693d..dfec3daca0a25 100644 --- a/task-sdk/src/airflow/sdk/importers/python_importer.py +++ b/task-sdk/src/airflow/sdk/importers/python_importer.py @@ -254,10 +254,9 @@ def _handle_sigsegv(signum, frame): if mod_name in sys.modules: del sys.modules[mod_name] - DagContext.current_autoregister_module_name = mod_name - def parse() -> list[types.ModuleType]: try: + DagContext.current_autoregister_module_name = mod_name with definition.import_context(): loader: importlib.abc.Loader if get_file_suffix(definition) == ".pyc": @@ -290,6 +289,10 @@ def parse() -> list[types.ModuleType]: ) ) return [] + finally: + # Reset on every exit: DagContext.pop() looks this name up in sys.modules, which may not + # hold it if the load did not complete. + DagContext.current_autoregister_module_name = None dagbag_import_timeout: float try: @@ -328,7 +331,6 @@ def _process_modules( } top_level_dags.update(DagContext.autoregistered_dags) - DagContext.current_autoregister_module_name = None DagContext.autoregistered_dags.clear() for dag, _mod in top_level_dags: From 80871c0e499537a19d6c2a3fa086ec41f6ed0631 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Wed, 28 Oct 2026 23:29:28 +0530 Subject: [PATCH 12/21] delete orphan dag_warning rows before recreating dag FK on downgrade --- .../migrations/versions/0152_3_5_0_add_dag_parse_warning.py | 1 + 1 file changed, 1 insertion(+) diff --git a/airflow-core/src/airflow/migrations/versions/0152_3_5_0_add_dag_parse_warning.py b/airflow-core/src/airflow/migrations/versions/0152_3_5_0_add_dag_parse_warning.py index 2891da20f717d..07109d9dd3335 100644 --- a/airflow-core/src/airflow/migrations/versions/0152_3_5_0_add_dag_parse_warning.py +++ b/airflow-core/src/airflow/migrations/versions/0152_3_5_0_add_dag_parse_warning.py @@ -46,6 +46,7 @@ def upgrade(): def downgrade(): + op.execute("DELETE FROM dag_warning WHERE dag_id NOT IN (SELECT dag_id FROM dag)") with op.batch_alter_table("dag_warning") as batch_op: batch_op.drop_column("filename") batch_op.drop_column("bundle_name") From 073e034520d2ff254286faf6302e5f57ed171996 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Thu, 29 Oct 2026 00:16:05 +0530 Subject: [PATCH 13/21] clear dag_warning rows in clear_db_dags test helper --- devel-common/src/tests_common/test_utils/db.py | 1 + 1 file changed, 1 insertion(+) diff --git a/devel-common/src/tests_common/test_utils/db.py b/devel-common/src/tests_common/test_utils/db.py index 221fd2cbb3b70..f4cab21b1e82f 100644 --- a/devel-common/src/tests_common/test_utils/db.py +++ b/devel-common/src/tests_common/test_utils/db.py @@ -322,6 +322,7 @@ def clear_db_dags(): session.execute(delete(DagFavorite)) session.execute(delete(DagTag)) session.execute(delete(DagOwnerAttributes)) + session.execute(delete(DagWarning)) session.execute( delete(DagRun) ) # todo: this should not be necessary because the fk to DagVersion should be ON DELETE SET NULL From 6eb6cd049133f53f167de24bedb41ff469878308 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Fri, 30 Oct 2026 23:05:35 +0530 Subject: [PATCH 14/21] skip file scoped dag warnings in stale purge; clear orphans on bundle refresh --- .../src/airflow/dag_processing/collection.py | 22 +++++++ .../src/airflow/dag_processing/manager.py | 17 +++++- airflow-core/src/airflow/models/dagwarning.py | 8 ++- .../tests/unit/models/test_dagwarning.py | 60 ++++++++++++++++++- 4 files changed, 103 insertions(+), 4 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/collection.py b/airflow-core/src/airflow/dag_processing/collection.py index 580d2b879f35d..f555b7c4d3424 100644 --- a/airflow-core/src/airflow/dag_processing/collection.py +++ b/airflow-core/src/airflow/dag_processing/collection.py @@ -417,6 +417,28 @@ def _update_file_scoped_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], diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index a8d0a6f3ce87e..5e020bbba8139 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/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. diff --git a/airflow-core/src/airflow/models/dagwarning.py b/airflow-core/src/airflow/models/dagwarning.py index a1ffbfda4cedf..035624ec9417d 100644 --- a/airflow-core/src/airflow/models/dagwarning.py +++ b/airflow-core/src/airflow/models/dagwarning.py @@ -99,9 +99,13 @@ def purge_inactive_dag_warnings(cls, *, session: Session = NEW_SESSION) -> None: """ if get_dialect_name(session) == "sqlite": dag_ids_stmt = select(DagModel.dag_id).where(DagModel.is_stale == true()) - query = delete(cls).where(cls.dag_id.in_(dag_ids_stmt.scalar_subquery())) + query = delete(cls).where( + cls.dag_id.in_(dag_ids_stmt.scalar_subquery()), cls.bundle_name.is_(None) + ) else: - query = delete(cls).where(cls.dag_id == DagModel.dag_id, DagModel.is_stale == true()) + query = delete(cls).where( + cls.dag_id == DagModel.dag_id, DagModel.is_stale == true(), cls.bundle_name.is_(None) + ) session.execute(query.execution_options(synchronize_session=False)) session.commit() diff --git a/airflow-core/tests/unit/models/test_dagwarning.py b/airflow-core/tests/unit/models/test_dagwarning.py index 0c0acdce8a45b..00b963b0a098a 100644 --- a/airflow-core/tests/unit/models/test_dagwarning.py +++ b/airflow-core/tests/unit/models/test_dagwarning.py @@ -24,7 +24,10 @@ from sqlalchemy import select from sqlalchemy.exc import OperationalError -from airflow.dag_processing.collection import _update_file_scoped_parse_failure_warnings +from airflow.dag_processing.collection import ( + _update_file_scoped_parse_failure_warnings, + clear_orphaned_file_scoped_parse_failure_warnings, +) from airflow.models import DagModel from airflow.models.dagwarning import ( PARSE_FAILURE_WARNING_TYPE, @@ -98,6 +101,61 @@ def test_purge_inactive_dag_warnings(self, session, testing_dag_bundle): assert len(remaining_dag_warnings) == 1 assert remaining_dag_warnings[0].dag_id == "dag_2" + def test_purge_inactive_dag_warnings_skips_file_scoped_parse_failures(self, session, testing_dag_bundle): + session.add( + DagModel(dag_id="stale_dag", bundle_name="testing", is_stale=True, relative_fileloc="factory.py") + ) + session.commit() + + session.add_all( + [ + DagWarning("stale_dag", DagWarningType.NONEXISTENT_POOL, "pool missing"), + DagWarning( + "stale_dag", + PARSE_FAILURE_WARNING_TYPE, + "partial factory failure", + bundle_name="testing", + filename="factory.py", + ), + ] + ) + session.commit() + + DagWarning.purge_inactive_dag_warnings(session=session) + + remaining = session.scalars(select(DagWarning)).all() + assert len(remaining) == 1 + assert remaining[0].warning_type == PARSE_FAILURE_WARNING_TYPE + + def test_clear_orphaned_file_scoped_parse_failure_warnings(self, session): + session.add_all( + [ + DagWarning( + "failed_a", + PARSE_FAILURE_WARNING_TYPE, + "boom", + bundle_name="testing", + filename="removed.py", + ), + DagWarning( + "failed_b", + PARSE_FAILURE_WARNING_TYPE, + "still relevant", + bundle_name="testing", + filename="still_here.py", + ), + ] + ) + session.commit() + + clear_orphaned_file_scoped_parse_failure_warnings("testing", {"still_here.py"}, session=session) + + session.commit() + + remaining = session.scalars(select(DagWarning)).all() + assert len(remaining) == 1 + assert remaining[0].filename == "still_here.py" + @mock.patch("airflow.models.dagwarning.delete") def test_retry_purge_inactive_dag_warnings(self, delete_mock): """ From 8e6b52b6214e77ed4b98a101d09b6381b43b2653 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Sat, 31 Oct 2026 13:48:57 +0530 Subject: [PATCH 15/21] fix static CI check failure --- .../src/airflow/sdk/definitions/_internal/contextmanager.py | 2 +- task-sdk/src/airflow/sdk/importers/base.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/task-sdk/src/airflow/sdk/definitions/_internal/contextmanager.py b/task-sdk/src/airflow/sdk/definitions/_internal/contextmanager.py index 05b3ff5374333..3d3fee5889983 100644 --- a/task-sdk/src/airflow/sdk/definitions/_internal/contextmanager.py +++ b/task-sdk/src/airflow/sdk/definitions/_internal/contextmanager.py @@ -23,11 +23,11 @@ from typing import TYPE_CHECKING, Any, Generic, TypeVar from airflow.sdk.definitions.dag import DAG -from airflow.sdk.definitions.parse_warning_record import DagParseWarningRecord from airflow.sdk.definitions.taskgroup import TaskGroup if TYPE_CHECKING: from airflow.sdk.definitions.context import Context + from airflow.sdk.definitions.parse_warning_record import DagParseWarningRecord T = TypeVar("T") diff --git a/task-sdk/src/airflow/sdk/importers/base.py b/task-sdk/src/airflow/sdk/importers/base.py index 4940b9eb8c55a..90bd8c1aa2ab1 100644 --- a/task-sdk/src/airflow/sdk/importers/base.py +++ b/task-sdk/src/airflow/sdk/importers/base.py @@ -29,7 +29,6 @@ from airflow.sdk._shared.module_loading.file_discovery import find_path_from_directory from airflow.sdk.configuration import conf -from airflow.sdk.definitions.parse_warning_record import DagParseWarningRecord from airflow.sdk.exceptions import AirflowConfigException from airflow.sdk.execution_time.coordinator import get_coordinator_manager @@ -39,6 +38,7 @@ from airflow.dag_processing.bundles.base import BaseDagBundle # noqa: SDK002 from airflow.sdk import DAG + from airflow.sdk.definitions.parse_warning_record import DagParseWarningRecord class _SiblingResource(Protocol): """A readable handle to a resource, e.g. a :class:`pathlib.Path` or :class:`zipfile.Path`.""" From c1fd6ea9be7f040286d5341ba157969537eaaecd Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Sat, 31 Oct 2026 14:20:46 +0530 Subject: [PATCH 16/21] fix React UI tests --- .../src/airflow/ui/src/pages/Dashboard/Stats/Stats.test.tsx | 1 + 1 file changed, 1 insertion(+) diff --git a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/Stats.test.tsx b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/Stats.test.tsx index 1289e2974f9a1..36d1f9d758cf8 100644 --- a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/Stats.test.tsx +++ b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/Stats.test.tsx @@ -53,6 +53,7 @@ vi.mock("src/utils", () => ({ useAutoRefresh: () => false, })); vi.mock("./DagImportErrors", () => ({ DagImportErrors: () => null })); +vi.mock("./ParseFailureWarnings", () => ({ ParseFailureWarnings: () => null })); vi.mock("./PluginImportErrors", () => ({ PluginImportErrors: () => null })); // Dashboard renders under the /home route. Mount Stats as that route's element so relative links From 6e51ebec7c73354566b8916aad599a6d64ec59a7 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Sat, 31 Oct 2026 16:47:45 +0530 Subject: [PATCH 17/21] fix unit tests CI - missing entries in public docs --- task-sdk/docs/api.rst | 2 ++ task-sdk/docs/index.rst | 1 + 2 files changed, 3 insertions(+) diff --git a/task-sdk/docs/api.rst b/task-sdk/docs/api.rst index 824fde2638e48..9e664fffa255f 100644 --- a/task-sdk/docs/api.rst +++ b/task-sdk/docs/api.rst @@ -150,6 +150,8 @@ Tasks & Operators .. autoapifunction:: airflow.sdk.get_parsing_context +.. autoapifunction:: airflow.sdk.emit_dag_warning + Retry Policies -------------- Retry policies let you configure per-exception retry behaviour as a parameter on any diff --git a/task-sdk/docs/index.rst b/task-sdk/docs/index.rst index c0653f6ee3fcf..5d6e2f627cc41 100644 --- a/task-sdk/docs/index.rst +++ b/task-sdk/docs/index.rst @@ -119,6 +119,7 @@ Why use ``airflow.sdk``? - :func:`airflow.sdk.cross_downstream` - :func:`airflow.sdk.get_current_context` - :func:`airflow.sdk.get_parsing_context` +- :func:`airflow.sdk.emit_dag_warning` All Dags must update their imports to refer to ``airflow.sdk`` instead of using internal Airflow modules directly. Deprecated legacy import paths, such as ``airflow.models.dag.DAG`` and ``airflow.decorator.task``, will be removed in a future version of Airflow. Some utilities and helper functions currently used from ``airflow.utils.*`` and other modules will gradually be migrated to the Task SDK over the next minor releases. These upcoming updates aim to completely separate Dag creation from internal Airflow services. Dag authors can look forward to continuous improvements to airflow.sdk, with no backwards-incompatible changes to their existing code. From 1edbb758b0ad438a304992953847b4db211c8b29 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Sat, 31 Oct 2026 20:35:10 +0530 Subject: [PATCH 18/21] update constants for failing CI checks --- airflow-core/tests/unit/dag_processing/test_manager.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py index 275422c78e459..9929cec902ccd 100644 --- a/airflow-core/tests/unit/dag_processing/test_manager.py +++ b/airflow-core/tests/unit/dag_processing/test_manager.py @@ -286,13 +286,14 @@ def _statement_breakdown(counts: Counter[tuple[str, str]]) -> str: # content is unchanged; once the hash has moved and [core] min_serialized_dag_update_interval has # lapsed it rewrites it, which costs two more statements per Dag and nothing extra per call. The # per-call price is a file that parsed cleanly: one reporting import errors also looks up whichever -# of them are already recorded. -FIXED_PER_CALL = 9 +# of them are already recorded. File-scoped parse-failure warnings add one DELETE on dag_warning per call. +FIXED_PER_CALL = 10 UNCHANGED_PER_DAG = 3 REWRITE_PER_DAG = 5 # A file that failed to parse and so defines no Dags. Two of the five are import_error SELECTs: the -# bounded lookup, and the listener re-reading the row the update beside it already had. -IMPORT_ERROR_PER_CALL = 5 +# bounded lookup, and the listener re-reading the row the update beside it already had. Plus one +# DELETE on dag_warning for the parsed file +IMPORT_ERROR_PER_CALL = 6 SWEEP_FILES = 4 # Calls the manager takes for that sweep: one per file today, 1 if a sweep is ever batched. From 93aabcae427f27f871148373ed10d9dc33ef2efa Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Sun, 1 Nov 2026 00:02:27 +0530 Subject: [PATCH 19/21] ci check failures fix --- task-sdk/docs/api.rst | 2 +- task-sdk/src/airflow/sdk/__init__.py | 2 +- task-sdk/src/airflow/sdk/__init__.pyi | 2 ++ 3 files changed, 4 insertions(+), 2 deletions(-) diff --git a/task-sdk/docs/api.rst b/task-sdk/docs/api.rst index 9e664fffa255f..afe4b933947de 100644 --- a/task-sdk/docs/api.rst +++ b/task-sdk/docs/api.rst @@ -384,7 +384,7 @@ Everything else .. autoapimodule:: airflow.sdk :members: :special-members: __version__ - :exclude-members: BaseAsyncOperator, BaseOperator, DAG, dag, asset, Asset, AssetAccessControl, AssetAlias, AssetAll, AssetAny, AssetWatcher, TaskGroup, TaskInstance, XComArg, get_current_context, get_parsing_context + :exclude-members: BaseAsyncOperator, BaseOperator, DAG, dag, asset, Asset, AssetAccessControl, AssetAlias, AssetAll, AssetAny, AssetWatcher, TaskGroup, TaskInstance, XComArg, get_current_context, get_parsing_context, emit_dag_warning :undoc-members: :imported-members: :no-index: diff --git a/task-sdk/src/airflow/sdk/__init__.py b/task-sdk/src/airflow/sdk/__init__.py index 077e19c2b46cd..c0cbf655132d9 100644 --- a/task-sdk/src/airflow/sdk/__init__.py +++ b/task-sdk/src/airflow/sdk/__init__.py @@ -170,6 +170,7 @@ from airflow.sdk.definitions.edges import EdgeModifier, Label from airflow.sdk.definitions.loop import LoopDecision from airflow.sdk.definitions.param import Param, ParamsDict + from airflow.sdk.definitions.parse_warning import emit_dag_warning from airflow.sdk.definitions.partition_mappers.allowed_key import AllowedKeyMapper from airflow.sdk.definitions.partition_mappers.base import ( PartitionMapper, @@ -278,7 +279,6 @@ "DeltaDataIntervalTimetable": (".definitions.timetables.interval", "DeltaDataIntervalTimetable"), "DeltaTriggerTimetable": (".definitions.timetables.trigger", "DeltaTriggerTimetable"), "EdgeModifier": (".definitions.edges", "EdgeModifier"), - "emit_dag_warning": (".definitions.parse_warning", "emit_dag_warning"), "EventsTimetable": (".definitions.timetables.events", "EventsTimetable"), "ExceptionRetryPolicy": (".definitions.retry_policy", "ExceptionRetryPolicy"), "FanOutMapper": (".definitions.partition_mappers.temporal", "FanOutMapper"), diff --git a/task-sdk/src/airflow/sdk/__init__.pyi b/task-sdk/src/airflow/sdk/__init__.pyi index 221d3eb48a6c6..e0040e78ad8a8 100644 --- a/task-sdk/src/airflow/sdk/__init__.pyi +++ b/task-sdk/src/airflow/sdk/__init__.pyi @@ -72,6 +72,7 @@ from airflow.sdk.definitions.decorators.task_group import task_group as task_gro from airflow.sdk.definitions.edges import EdgeModifier as EdgeModifier, Label as Label from airflow.sdk.definitions.loop import LoopDecision as LoopDecision from airflow.sdk.definitions.param import Param as Param +from airflow.sdk.definitions.parse_warning import emit_dag_warning as emit_dag_warning from airflow.sdk.definitions.partition_mappers.allowed_key import AllowedKeyMapper from airflow.sdk.definitions.partition_mappers.base import PartitionMapper, RollupMapper from airflow.sdk.definitions.partition_mappers.chain import ChainMapper @@ -176,6 +177,7 @@ __all__ = [ "DeltaDataIntervalTimetable", "DeltaTriggerTimetable", "EdgeModifier", + "emit_dag_warning", "EventsTimetable", "ExceptionRetryPolicy", "FanOutMapper", From 4f064e4d381217bb6d1c63fe2429b0c171baacc5 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Sun, 1 Nov 2026 00:20:56 +0530 Subject: [PATCH 20/21] revert unwanted change --- task-sdk/src/airflow/sdk/__init__.py | 1 + 1 file changed, 1 insertion(+) diff --git a/task-sdk/src/airflow/sdk/__init__.py b/task-sdk/src/airflow/sdk/__init__.py index c0cbf655132d9..832360be62b77 100644 --- a/task-sdk/src/airflow/sdk/__init__.py +++ b/task-sdk/src/airflow/sdk/__init__.py @@ -279,6 +279,7 @@ "DeltaDataIntervalTimetable": (".definitions.timetables.interval", "DeltaDataIntervalTimetable"), "DeltaTriggerTimetable": (".definitions.timetables.trigger", "DeltaTriggerTimetable"), "EdgeModifier": (".definitions.edges", "EdgeModifier"), + "emit_dag_warning": (".definitions.parse_warning", "emit_dag_warning"), "EventsTimetable": (".definitions.timetables.events", "EventsTimetable"), "ExceptionRetryPolicy": (".definitions.retry_policy", "ExceptionRetryPolicy"), "FanOutMapper": (".definitions.partition_mappers.temporal", "FanOutMapper"), From 575d4b25345cfbf27a3f9ce83dca3a5f0fbeb9a8 Mon Sep 17 00:00:00 2001 From: ydsakshi Date: Mon, 2 Nov 2026 00:02:43 +0530 Subject: [PATCH 21/21] rename, import refactor --- airflow-core/src/airflow/dag_processing/collection.py | 4 +++- airflow-core/src/airflow/dag_processing/dagbag.py | 4 ++-- .../airflow/sdk/execution_time/schema/versions/v2026_10_30.py | 2 ++ task-sdk/src/airflow/sdk/importers/python_importer.py | 2 +- 4 files changed, 8 insertions(+), 4 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/collection.py b/airflow-core/src/airflow/dag_processing/collection.py index f555b7c4d3424..7e61da600b167 100644 --- a/airflow-core/src/airflow/dag_processing/collection.py +++ b/airflow-core/src/airflow/dag_processing/collection.py @@ -387,7 +387,9 @@ def _update_dag_warnings( session.merge(warning_to_add) -def dag_parse_failure_warnings(bundle_name: str, warnings_by_rel_path: dict[str, list]) -> list[DagWarning]: +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, diff --git a/airflow-core/src/airflow/dag_processing/dagbag.py b/airflow-core/src/airflow/dag_processing/dagbag.py index 23943c923f176..1f7ac8ffb7207 100644 --- a/airflow-core/src/airflow/dag_processing/dagbag.py +++ b/airflow-core/src/airflow/dag_processing/dagbag.py @@ -707,9 +707,9 @@ 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 dag_parse_failure_warnings + from airflow.dag_processing.collection import build_dag_parse_failure_warnings - parse_failure_warnings = dag_parse_failure_warnings(bundle_name, dagbag.parse_warnings) + parse_failure_warnings = build_dag_parse_failure_warnings(bundle_name, dagbag.parse_warnings) update_dag_parsing_results_in_db( bundle_name, diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py index aaa0b2d8cffb8..fddc8f3bca776 100644 --- a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py +++ b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py @@ -106,6 +106,8 @@ class AddDagDefinitionsToDagFileParsingResult(VersionChange): class AddParseWarningsToDagFileParsingResult(VersionChange): + """Add parse-time DAG warnings collected while importing a DAG file to `DagFileParsingResult`.""" + description = __doc__ instructions_to_migrate_to_previous_version = ( schema(DagFileParsingResult).field("parse_warnings").didnt_exist, diff --git a/task-sdk/src/airflow/sdk/importers/python_importer.py b/task-sdk/src/airflow/sdk/importers/python_importer.py index dfec3daca0a25..05b9348fad94a 100644 --- a/task-sdk/src/airflow/sdk/importers/python_importer.py +++ b/task-sdk/src/airflow/sdk/importers/python_importer.py @@ -36,6 +36,7 @@ from airflow.sdk.configuration import conf from airflow.sdk.definitions._internal.contextmanager import DagContext from airflow.sdk.definitions.dag import DAG +from airflow.sdk.definitions.parse_warning import drain_parse_warnings from airflow.sdk.exceptions import AirflowConfigException from airflow.sdk.execution_time.timeout import timeout from airflow.sdk.importers.base import ( @@ -212,7 +213,6 @@ def import_definition( ) self._process_modules(modules, result, bundle=bundle) - from airflow.sdk.definitions.parse_warning import drain_parse_warnings result.parse_warnings.extend(drain_parse_warnings()) return result