This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 098d0174731 Report a Dag whose timetable fails to load as an import
error (#74709)
098d0174731 is described below
commit 098d0174731a64f59f929d53bf1f8c8ca87601ca
Author: Adam Postek <[email protected]>
AuthorDate: Thu Oct 15 15:42:47 2026 +0000
Report a Dag whose timetable fails to load as an import error (#74709)
---
airflow-core/newsfragments/74709.bugfix.rst | 1 +
.../src/airflow/dag_processing/collection.py | 27 +++++++++++++
.../tests/unit/dag_processing/test_collection.py | 44 ++++++++++++++++++++++
3 files changed, 72 insertions(+)
diff --git a/airflow-core/newsfragments/74709.bugfix.rst
b/airflow-core/newsfragments/74709.bugfix.rst
new file mode 100644
index 00000000000..b8deba9e0d2
--- /dev/null
+++ b/airflow-core/newsfragments/74709.bugfix.rst
@@ -0,0 +1 @@
+Report a Dag whose timetable fails to load, for example because a custom
timetable reads a missing Variable, as an import error in the UI. Previously
the error stopped every Dag in the same file from being saved and was only
visible in the Dag processor log.
diff --git a/airflow-core/src/airflow/dag_processing/collection.py
b/airflow-core/src/airflow/dag_processing/collection.py
index cd7228d210f..70451890050 100644
--- a/airflow-core/src/airflow/dag_processing/collection.py
+++ b/airflow-core/src/airflow/dag_processing/collection.py
@@ -573,6 +573,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,
@@ -616,6 +642,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/tests/unit/dag_processing/test_collection.py
b/airflow-core/tests/unit/dag_processing/test_collection.py
index 6e9a61c8756..4eec804683f 100644
--- a/airflow-core/tests/unit/dag_processing/test_collection.py
+++ b/airflow-core/tests/unit/dag_processing/test_collection.py
@@ -84,6 +84,7 @@ from airflow.sdk import (
from airflow.sdk.definitions.deadline import AsyncCallback,
BaseDeadlineReference, DeadlineAlert
from airflow.sdk.definitions.timetables.assets import AssetOrTimeSchedule,
PartitionedAssetTimetable
from airflow.sdk.importers import DagSourceCode
+from airflow.serialization.decoders import decode_timetable
from airflow.serialization.definitions.assets import SerializedAsset
from airflow.serialization.encoders import encode_trigger,
ensure_serialized_asset
from airflow.serialization.serialized_objects import LazyDeserializedDAG
@@ -1094,6 +1095,49 @@ class TestUpdateDagParsingResults:
assert len(dag_import_error_listener.existing) == 0
assert dag_import_error_listener.new["abc.py"] ==
import_error.stacktrace
+ @patch.object(ParseImportError, "full_file_path")
+ @pytest.mark.usefixtures("clean_db")
+ def test_timetable_that_fails_to_load_is_an_import_error(
+ self, mock_full_path, session, testing_dag_bundle
+ ):
+ """
+ A timetable that raises when rebuilt (for example a custom timetable
reading a missing Variable
+ in ``__init__``) is reported as an import error for its file, and only
that Dag is skipped.
+ """
+ mock_full_path.return_value = "abc.py"
+ dags = []
+ for dag_id, cron in (("good_dag", "0 0 * * *"), ("bad_dag", "0 1 * *
*")):
+ dag = DAG(dag_id=dag_id, schedule=CronTriggerTimetable(cron,
timezone="UTC"))
+ dag.fileloc = dag.relative_fileloc = "abc.py"
+ dags.append(LazyDeserializedDAG.from_dag(dag))
+
+ def fail_for_bad_dag(var):
+ if var["__var"]["expression"] == "0 1 * * *":
+ raise RuntimeError("Variable something not found")
+ return decode_timetable(var)
+
+ import_errors: dict[tuple[str, str], str] = {}
+ with
patch("airflow.serialization.serialized_objects.decode_timetable",
side_effect=fail_for_bad_dag):
+ update_dag_parsing_results_in_db(
+ "testing",
+ None,
+ dags,
+ import_errors,
+ None,
+ set(),
+ session,
+ files_parsed={("testing", "abc.py")},
+ )
+
+ assert session.get(DagModel, ("good_dag",)) is not None
+ assert session.get(DagModel, ("bad_dag",)) is None
+ error = import_errors[("testing", "abc.py")]
+ assert "bad_dag" in error
+ assert "Traceback (most recent call last)" in error
+ assert "RuntimeError: Variable something not found" in error
+ [stored] = session.scalars(select(ParseImportError)).all()
+ assert stored.filename == "abc.py"
+
@patch.object(ParseImportError, "full_file_path")
@mark_fab_auth_manager_test
@conf_vars({("core", "min_serialized_dag_update_interval"): "5"})
