This is an automated email from the ASF dual-hosted git repository.
shahar1 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 7b5fc4b7068 Reset Dag autoregistration state when a Python Dag file
fails to load (#74911)
7b5fc4b7068 is described below
commit 7b5fc4b70688c1bfd612eaec3ad5346c85319303
Author: Shahar Epstein <[email protected]>
AuthorDate: Sat Oct 24 22:28:09 2026 +0300
Reset Dag autoregistration state when a Python Dag file fails to load
(#74911)
In parallel test runs, a failed Python Dag import could leave stale state
that broke later imports in the same worker; the fix always clears it.
---
.../src/airflow/sdk/importers/python_importer.py | 8 +++--
.../task_sdk/importers/test_python_importer.py | 37 ++++++++++++++++++++++
2 files changed, 42 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 e6fae5b4452..502745792f2 100644
--- a/task-sdk/src/airflow/sdk/importers/python_importer.py
+++ b/task-sdk/src/airflow/sdk/importers/python_importer.py
@@ -250,10 +250,9 @@ class
PythonDagImporter(AbstractDagImporter[FileDagDefinition]):
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":
@@ -285,6 +284,10 @@ class
PythonDagImporter(AbstractDagImporter[FileDagDefinition]):
)
)
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:
@@ -323,7 +326,6 @@ class
PythonDagImporter(AbstractDagImporter[FileDagDefinition]):
}
top_level_dags.update(DagContext.autoregistered_dags)
- DagContext.current_autoregister_module_name = None
DagContext.autoregistered_dags.clear()
for dag, _mod in top_level_dags:
diff --git a/task-sdk/tests/task_sdk/importers/test_python_importer.py
b/task-sdk/tests/task_sdk/importers/test_python_importer.py
index 54cfdc94df6..eebbf861916 100644
--- a/task-sdk/tests/task_sdk/importers/test_python_importer.py
+++ b/task-sdk/tests/task_sdk/importers/test_python_importer.py
@@ -33,6 +33,7 @@ from unittest import mock
import pytest
+from airflow.sdk import DAG
from airflow.sdk.exceptions import AirflowConfigException
from airflow.sdk.importers import (
DagImportError,
@@ -347,6 +348,42 @@ class TestPythonDagImporter:
bundle=mock_bundle,
)
+ @pytest.mark.parametrize(
+ ("timeout_policy", "expectation"),
+ [
+ pytest.param(
+ {"return_value": "invalid_timeout_str"},
+ pytest.raises(AirflowConfigException),
+ id="config-error-propagates",
+ ),
+ pytest.param(
+ {"side_effect": RuntimeError("policy failed")},
+ contextlib.nullcontext(),
+ id="error-reported",
+ ),
+ pytest.param(
+ {"return_value": 0},
+ pytest.raises(KeyboardInterrupt),
+ id="keyboard-interrupt-propagates",
+ ),
+ ],
+ )
+ def test_failed_load_does_not_break_later_dags(self, mock_bundle,
timeout_policy, expectation):
+ (mock_bundle.path / "dag.py").write_text("raise KeyboardInterrupt\n")
+ mock_settings = mock.MagicMock(spec=["get_dagbag_import_timeout"])
+
mock_settings.get_dagbag_import_timeout.configure_mock(**timeout_policy)
+ mock_airflow = mock.MagicMock(spec=["settings"],
settings=mock_settings)
+
+ with mock.patch.dict("sys.modules", {"airflow": mock_airflow}),
expectation:
+ PythonDagImporter().import_definition(
+ FilesystemDagDefinition(path=mock_bundle.path / "dag.py"),
+ bundle=mock_bundle,
+ )
+
+ with DAG("built_after_failed_load") as dag:
+ pass
+ assert dag.dag_id == "built_after_failed_load"
+
@mock.patch.object(PythonDagImporter, "_load_modules",
side_effect=TypeError("unexpected None"))
def test_unexpected_type_error_captured_in_result_errors(self, mock_load,
mock_bundle):
importer = PythonDagImporter()
