This is an automated email from the ASF dual-hosted git repository. potiuk 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 addbd587e2 Allow email field to be templated (#35546) addbd587e2 is described below commit addbd587e218a3ace09b08852094b336ecb33e9b Author: Gevorg Davoian <gdavo...@gmail.com> AuthorDate: Thu Nov 9 20:09:42 2023 +0200 Allow email field to be templated (#35546) --- airflow/serialization/serialized_objects.py | 4 +++- tests/serialization/test_dag_serialization.py | 29 +++++++++++++++++++-------- 2 files changed, 24 insertions(+), 9 deletions(-) diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py index ef35d843b4..d2a72cb1ca 100644 --- a/airflow/serialization/serialized_objects.py +++ b/airflow/serialization/serialized_objects.py @@ -878,10 +878,12 @@ class SerializedBaseOperator(BaseOperator, BaseSerialization): # 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()) + # Though allow some of the BaseOperator fields to be templated anyway + forbidden_fields.difference_update({"email"}) 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}") + raise AirflowException(f"Cannot template BaseOperator field: {template_field!r}") 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 4aecebcd1d..a9c63c80fe 100644 --- a/tests/serialization/test_dag_serialization.py +++ b/tests/serialization/test_dag_serialization.py @@ -2035,20 +2035,33 @@ class TestStringifiedDAGs: assert param.description == "hello" assert param.schema == {"type": "string"} - def test_not_templateable_fields_in_serialized_dag( - self, - ): + @pytest.mark.db_test + def test_not_templateable_fields_in_serialized_dag(self): """ - Test that when we use not templateable fields, an Airflow exception is raised. + Test that when we use not templateable fields, an Airflow exception is raised. """ class TestOperator(BaseOperator): - template_fields = ("execution_timeout",) + template_fields = ( + "email", # templateable + "execution_timeout", # not templateable + ) + + def execute(self, context: Context): + pass + + dag = DAG(dag_id="test_dag", start_date=datetime(2023, 11, 9)) - 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"): + task = TestOperator( + task_id="test_task", + email="{{ ','.join(test_email_list) }}", + execution_timeout=timedelta(seconds=10), + ) + task.render_template_fields(context={"test_email_list": ["f...@test.com", "b...@test.com"]}) + assert task.email == "f...@test.com,b...@test.com" + + with pytest.raises(AirflowException, match="Cannot template BaseOperator field: 'execution_timeout'"): SerializedDAG.to_dict(dag)