Add a check for not templateable fields

airflow/serialization/serialized_objects.py

@@ -20,6 +20,7 @@
 import collections.abc
 import datetime
 import enum
+import inspect
 import logging
 import warnings
 import weakref
@@ -702,6 +703,7 @@ def __init__(self, *args, **kwargs):
     def task_type(self) -> str:
         # Overwrites task_type of BaseOperator to use _task_type instead of
         # __class__.__name__.
+
         return self._task_type
 
     @task_type.setter
@@ -770,8 +772,12 @@ def _serialize_node(cls, op: BaseOperator | MappedOperator, include_deps: bool)
 
         # Store all template_fields as they are if there are JSON Serializable
         # If not, store them as strings
+        # And raise an exception if the field is not templateable
+        forbidden_fields = set(inspect.signature(BaseOperator.__init__).parameters.keys())
         if op.template_fields:
             for template_field in op.template_fields:
+                if template_field in forbidden_fields:
+                    raise AirflowException(f"Cannot template BaseOperator fields: {template_field}")
                 value = getattr(op, template_field, None)
                 if not cls._is_excluded(value, template_field, op):
                     serialize_op[template_field] = serialize_template_field(value)

tests/serialization/test_dag_serialization.py

@@ -38,7 +38,7 @@
 
 import airflow
 from airflow.datasets import Dataset
-from airflow.exceptions import SerializationError
+from airflow.exceptions import AirflowException, SerializationError
 from airflow.hooks.base import BaseHook
 from airflow.kubernetes.pod_generator import PodGenerator
 from airflow.models import DAG, Connection, DagBag, Operator
@@ -2016,6 +2016,22 @@ def test_params_serialize_default(self):
         assert param.description == "hello"
         assert param.schema == {"type": "string"}
 
+    def test_not_templateable_fields_in_serialized_dag(
+        self,
+    ):
+        """
+        Test that when we use  not templateable fields, an Airflow exception is raised.
+        """
+
+        class TestOperator(BaseOperator):
+            template_fields = ("execution_timeout",)
+
+        dag = DAG("test_not_templateable_fields", start_date=datetime(2019, 8, 1))
+        with dag:
+            TestOperator(task_id="test", execution_timeout=timedelta(seconds=10))
+        with pytest.raises(AirflowException, match="Cannot template BaseOperator fields: execution_timeout"):
+            SerializedDAG.to_dict(dag)
+
 
 def test_kubernetes_optional():
     """Serialisation / deserialisation continues to work without kubernetes installed"""