diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py index db20b61d747ca..7200094547f0a 100644 --- a/airflow/serialization/serialized_objects.py +++ b/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) diff --git a/tests/serialization/test_dag_serialization.py b/tests/serialization/test_dag_serialization.py index 43ce80c09c075..967c433bdc160 100644 --- a/tests/serialization/test_dag_serialization.py +++ b/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"""