This is an automated email from the ASF dual-hosted git repository.
shahar1 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 52e27183bf7 Read HITLOperator subject/body into the summary lazily
(#70345)
52e27183bf7 is described below
commit 52e27183bf761536e1df526364311337ca357406
Author: Stefan Wang <[email protected]>
AuthorDate: Fri Jul 24 02:15:48 2026 -0700
Read HITLOperator subject/body into the summary lazily (#70345)
* Read HITLOperator subject/body into the summary lazily
subject and body are template fields, rendered after __init__ runs. The
constructor built hitl_summary with self.subject/self.body, so a templated
subject or body was captured as its un-rendered Jinja expression -- and the
OpenLineage task START event, which reads hitl_summary, emitted the raw
template instead of the rendered value.
Make hitl_summary a property so subject/body are read at access time, after
rendering. Runtime and subclass additions go through a private
_hitl_summary_extra dict that the property merges in.
related: #70296
---
.../airflow/providers/standard/operators/hitl.py | 33 +++++++++++++++-------
.../tests/unit/standard/operators/test_hitl.py | 18 +++++++++++-
.../ci/prek/validate_operators_init_exemptions.txt | 1 -
3 files changed, 40 insertions(+), 12 deletions(-)
diff --git
a/providers/standard/src/airflow/providers/standard/operators/hitl.py
b/providers/standard/src/airflow/providers/standard/operators/hitl.py
index ce42221e7d9..ce4f4d4a878 100644
--- a/providers/standard/src/airflow/providers/standard/operators/hitl.py
+++ b/providers/standard/src/airflow/providers/standard/operators/hitl.py
@@ -132,8 +132,18 @@ class HITLOperator(BaseOperator):
self.validate_params()
self.validate_defaults()
- # HITL summary for the use of listeners; subclasses can extend it.
- self.hitl_summary: dict[str, Any] = {
+ # Runtime/subclass additions to the summary; config-derived entries
live in the property.
+ self._hitl_summary_extra: dict[str, Any] = {}
+
+ @property
+ def hitl_summary(self) -> dict[str, Any]:
+ """
+ Summary of the Human-in-the-loop request, for listeners/observability.
+
+ A property so the ``subject``/``body`` template fields are read after
rendering, not
+ captured as un-rendered Jinja in ``__init__``.
+ """
+ return {
"subject": self.subject,
"body": self.body,
"options": self.options,
@@ -141,6 +151,7 @@ class HITLOperator(BaseOperator):
"multiple": self.multiple,
"assigned_users": self.assigned_users,
"serialized_params": self.serialized_params or None,
+ **self._hitl_summary_extra,
}
def validate_options(self) -> None:
@@ -210,7 +221,9 @@ class HITLOperator(BaseOperator):
timeout_datetime = None
# Enrich summary with runtime info
- self.hitl_summary["timeout_datetime"] = timeout_datetime.isoformat()
if timeout_datetime else None
+ self._hitl_summary_extra["timeout_datetime"] = (
+ timeout_datetime.isoformat() if timeout_datetime else None
+ )
self.log.info("Waiting for response")
for notifier in self.notifiers:
@@ -246,7 +259,7 @@ class HITLOperator(BaseOperator):
def execute_complete(self, context: Context, event: dict[str, Any]) -> Any:
if "error" in event:
- self.hitl_summary["error_type"] = event["error_type"]
+ self._hitl_summary_extra["error_type"] = event["error_type"]
self.process_trigger_event_error(event)
chosen_options = event["chosen_options"]
@@ -254,7 +267,7 @@ class HITLOperator(BaseOperator):
self.validate_chosen_options(chosen_options)
self.validate_params_input(params_input)
- self.hitl_summary.update(
+ self._hitl_summary_extra.update(
{
"chosen_options": chosen_options,
"params_input": params_input,
@@ -432,14 +445,14 @@ class ApprovalOperator(HITLOperator, SkipMixin):
**kwargs,
)
- self.hitl_summary["ignore_downstream_trigger_rules"] =
self.ignore_downstream_trigger_rules
- self.hitl_summary["fail_on_reject"] = self.fail_on_reject
+ self._hitl_summary_extra["ignore_downstream_trigger_rules"] =
self.ignore_downstream_trigger_rules
+ self._hitl_summary_extra["fail_on_reject"] = self.fail_on_reject
def execute_complete(self, context: Context, event: dict[str, Any]) -> Any:
ret = super().execute_complete(context=context, event=event)
chosen_option = ret["chosen_options"][0]
- self.hitl_summary["approved"] = chosen_option == self.APPROVE
+ self._hitl_summary_extra["approved"] = chosen_option == self.APPROVE
if chosen_option == self.APPROVE:
self.log.info("Approved. Proceeding with downstream tasks...")
return ret
@@ -493,7 +506,7 @@ class HITLBranchOperator(HITLOperator, BranchMixIn):
super().__init__(**kwargs)
self.options_mapping = options_mapping or {}
self.validate_options_mapping()
- self.hitl_summary["options_mapping"] = self.options_mapping
+ self._hitl_summary_extra["options_mapping"] = self.options_mapping
def validate_options_mapping(self) -> None:
"""
@@ -528,7 +541,7 @@ class HITLBranchOperator(HITLOperator, BranchMixIn):
# Map options to task IDs using the mapping, fallback to original
option
chosen_options = [self.options_mapping.get(option, option) for option
in chosen_options]
- self.hitl_summary["branches_to_execute"] = chosen_options
+ self._hitl_summary_extra["branches_to_execute"] = chosen_options
return self.do_branch(context=context,
branches_to_execute=chosen_options)
diff --git a/providers/standard/tests/unit/standard/operators/test_hitl.py
b/providers/standard/tests/unit/standard/operators/test_hitl.py
index 7b7d5941b60..e83d3c91446 100644
--- a/providers/standard/tests/unit/standard/operators/test_hitl.py
+++ b/providers/standard/tests/unit/standard/operators/test_hitl.py
@@ -974,6 +974,21 @@ class TestHITLSummaryForListeners:
"serialized_params": None,
}
+ def test_summary_reflects_rendered_subject_body(self) -> None:
+ """The summary reads subject/body live, so it reflects rendered values
(guards #70296)."""
+ op = HITLOperator(
+ task_id="test",
+ subject="Review for {{ ds }}",
+ body="Deploy {{ ds }}?",
+ options=["Yes", "No"],
+ )
+ # Airflow renders template fields in place before execute.
+ op.subject = "Review for 2020-01-01"
+ op.body = "Deploy 2020-01-01?"
+
+ assert op.hitl_summary["subject"] == "Review for 2020-01-01"
+ assert op.hitl_summary["body"] == "Deploy 2020-01-01?"
+
def test_approval_operator_init_summary(self) -> None:
"""ApprovalOperator hitl_summary includes base + approval-specific
fields."""
op = ApprovalOperator(
@@ -1382,7 +1397,8 @@ class TestHITLSummaryForListeners:
},
)
- assert s == {
+ # hitl_summary is a property, so re-read it to see the
execute_complete additions.
+ assert op.hitl_summary == {
"subject": "Release v2.0?",
"body": "Please approve the production deployment.",
"options": ["Approve", "Reject"],
diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt
b/scripts/ci/prek/validate_operators_init_exemptions.txt
index 7f5f7bc618c..a4846a7da2e 100644
--- a/scripts/ci/prek/validate_operators_init_exemptions.txt
+++ b/scripts/ci/prek/validate_operators_init_exemptions.txt
@@ -68,7 +68,6 @@
providers/papermill/src/airflow/providers/papermill/operators/papermill.py::Pape
providers/ssh/src/airflow/providers/ssh/operators/ssh.py::SSHOperator
providers/ssh/src/airflow/providers/ssh/operators/ssh_remote_job.py::SSHRemoteJobOperator
providers/standard/src/airflow/providers/standard/operators/bash.py::BashOperator
-providers/standard/src/airflow/providers/standard/operators/hitl.py::HITLOperator
providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py::TriggerDagRunOperator
providers/standard/src/airflow/providers/standard/sensors/date_time.py::DateTimeSensor
providers/teradata/src/airflow/providers/teradata/transfers/teradata_to_teradata.py::TeradataToTeradataOperator
