Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Allow email field to be templated #35546

Merged
merged 4 commits into from
Nov 9, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion airflow/serialization/serialized_objects.py
Original file line number Diff line number Diff line change
Expand Up @@ -878,10 +878,12 @@ def _serialize_node(cls, op: BaseOperator | MappedOperator, include_deps: bool)
# 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)
Expand Down
29 changes: 21 additions & 8 deletions tests/serialization/test_dag_serialization.py
Original file line number Diff line number Diff line change
Expand Up @@ -2035,20 +2035,33 @@ def test_params_serialize_default(self):
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": ["foo@test.com", "bar@test.com"]})
assert task.email == "foo@test.com,bar@test.com"

with pytest.raises(AirflowException, match="Cannot template BaseOperator field: 'execution_timeout'"):
SerializedDAG.to_dict(dag)
gdavoian marked this conversation as resolved.
Show resolved Hide resolved


Expand Down