Skip the Python-only parse checks for non-Python Dag files

The Airflow module pre-import and the Dag version inflation check read a
Dag file as Python source, which fails for a binary file such as a JAR.
Both now run only for .py files.

Author: jason810496

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

@@ -76,7 +76,11 @@
 from airflow.sdk.importers import DagSourceCode  # noqa: TC001
 from airflow.serialization.definitions.dag import SerializedDAG
 from airflow.serialization.serialized_objects import DagSerialization, LazyDeserializedDAG
-from airflow.utils.dag_version_inflation_checker import check_dag_file_stability
+from airflow.utils.dag_version_inflation_checker import (
+    DagVersionInflationCheckLevel,
+    DagVersionInflationCheckResult,
+    check_dag_file_stability,
+)
 from airflow.utils.file import iter_airflow_imports
 from airflow.utils.helpers import prune_dict
 from airflow.utils.log.logging_mixin import LoggingMixin
@@ -175,6 +179,11 @@ class DagFileParsingResult(BaseModel):
 ]
 
 
+def _is_python_source(file_path: str | os.PathLike[str]) -> bool:
+    """Return whether a Dag file is Python source, which the AST-based pre-parse steps can read."""
+    return Path(file_path).suffix.lower() == ".py"
+
+
 def _pre_import_airflow_modules(file_path: str, log: FilteringBoundLogger) -> None:
     """
     Pre-import Airflow modules found in the given file.
@@ -186,6 +195,8 @@ def _pre_import_airflow_modules(file_path: str, log: FilteringBoundLogger) -> No
     :param file_path: Path to the file to scan for imports
     :param log: Logger instance to use for warnings
     """
+    if not _is_python_source(file_path):
+        return
     if not conf.getboolean("dag_processor", "parsing_pre_import_modules", fallback=True):
         return
 
@@ -226,7 +237,11 @@ def _parse_file_entrypoint():
 def _parse_file(msg: DagFileParseRequest, log: FilteringBoundLogger) -> DagFileParsingResult | None:
     # TODO: Set known_pool names on DagBag!
 
-    stability_check_result = check_dag_file_stability(os.fspath(msg.file))
+    stability_check_result = (
+        check_dag_file_stability(os.fspath(msg.file))
+        if _is_python_source(msg.file)
+        else DagVersionInflationCheckResult(check_level=DagVersionInflationCheckLevel.off)
+    )
 
     # Callback runs must not be blocked by the stability check: callbacks for
     # already-scheduled runs still have to execute, and they never produce a

airflow-core/tests/unit/dag_processing/test_processor.py

@@ -94,6 +94,7 @@
 from airflow.utils.state import TaskInstanceState
 
 from tests_common.test_utils.config import conf_vars, env_vars
+from unit.dag_processing.fake_importers import FAKE_IMPORTER, task_sdk_importers
 
 if TYPE_CHECKING:
     from kgb import SpyAgency
@@ -433,6 +434,16 @@ def test__pre_import_airflow_modules_when_disabled(self):
         mock_iter.assert_not_called()
         logger.warning.assert_not_called()
 
+    def test__pre_import_airflow_modules_skips_non_python_files(self):
+        logger = MagicMock(spec=FilteringBoundLogger)
+        with (
+            env_vars({"AIRFLOW__DAG_PROCESSOR__PARSING_PRE_IMPORT_MODULES": "true"}),
+            patch("airflow.dag_processing.processor.iter_airflow_imports") as mock_iter,
+        ):
+            _pre_import_airflow_modules("bundle.jar", logger)
+
+        mock_iter.assert_not_called()
+
     def test__pre_import_airflow_modules_when_enabled(self):
         logger = MagicMock(spec=FilteringBoundLogger)
         with (
@@ -695,6 +706,26 @@ def fake_collect_dags(self, *args, **kwargs):
     assert called is True
 
 
+def test_parse_file_skips_python_checks_for_other_importers(tmp_path):
+    native = tmp_path / "native.fake"
+    native.write_text("native_dag\n")
+
+    with (
+        task_sdk_importers(FAKE_IMPORTER),
+        patch("airflow.dag_processing.processor.check_dag_file_stability", autospec=True) as stability_check,
+    ):
+        result = _parse_file(
+            DagFileParseRequest(file=str(native), bundle_path=tmp_path, bundle_name="testing"),
+            log=structlog.get_logger(),
+        )
+
+    assert result is not None
+    assert result.import_errors == {}
+    assert [dag.dag_id for dag in result.serialized_dags] == ["native_dag"]
+    assert result.serialized_dags[0].data["dag"]["relative_fileloc"] == "native.fake"
+    stability_check.assert_not_called()
+
+
 @conf_vars({("dag_processor", "dag_version_inflation_check_level"): "error"})
 def test_parse_file_static_check_with_error():
     result = _parse_file(