This is an automated email from the ASF dual-hosted git repository.
ashb pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new d7e00b5084f [v3-4-test] Report a Dag whose timetable fails to load as
an import error (#74709) (#74807)
d7e00b5084f is described below
commit d7e00b5084f9df89a2a68fa21bcbb6d2ae9c5700
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Oct 16 17:12:57 2026 +0100
[v3-4-test] Report a Dag whose timetable fails to load as an import error
(#74709) (#74807)
(cherry picked from commit 098d0174731a64f59f929d53bf1f8c8ca87601ca)
Co-authored-by: Adam Postek <[email protected]>
---
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"})
