kaxil commented on code in PR #44843:
URL: https://github.com/apache/airflow/pull/44843#discussion_r1882836965


##########
task_sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -157,22 +157,26 @@ def startup() -> tuple[RuntimeTaskInstance, Logger]:
     # 1. Implementing the part where we pull in the logic to render fields and 
add that here
     # for all operators, we should do setattr(task, templated_field, 
rendered_templated_field)
     # task.templated_fields should give all the templated_fields and each of 
those fields should
-    # give the rendered values.
+    # give the rendered values. task.templated_fields should already be in a 
JSONable format and
+    # we should not have to handle that here.
 
     # 2. Once rendered, we call the `set_rtif` API to store the rtif in the 
metadata DB
-    templated_fields = ti.task.template_fields
-    payload = {}
-
-    for field in templated_fields:
-        if field not in payload:
-            payload[field] = getattr(ti.task, field)
 
     # so that we do not call the API unnecessarily
-    if payload:
-        SUPERVISOR_COMMS.send_request(log=log, 
msg=SetRenderedFields(rendered_fields=payload))
+    if rendered_fields := _get_rendered_fields(ti.task):
+        SUPERVISOR_COMMS.send_request(log=log, 
msg=SetRenderedFields(rendered_fields=rendered_fields))
     return ti, log
 
 
+def _get_rendered_fields(task: BaseOperator) -> dict[str, JsonValue]:
+    # TODO: Port one of the following to Task SDK
+    #   airflow.serialization.helpers.serialize_template_field or
+    #   airflow.models.renderedtifields.get_serialized_template_fields
+    from airflow.serialization.helpers import serialize_template_field
+
+    return {field: serialize_template_field(getattr(task, field), field) for 
field in task.template_fields}

Review Comment:
   If we don't use `serialize_template_field`, we will have to duplicate the 
logic -- which is also fine -- but we need to take an informed call. On one 
hand we need to de-couple, on the other hand we can keep it centralized and 
port it later
   
   ---
   
   There is also logic to redact:
   
   
https://github.com/apache/airflow/blob/cc9b9a8794da50630e26f5de04b4c0ae5d49353f/airflow/serialization/helpers.py#L52
   
   
https://github.com/apache/airflow/blob/cc9b9a8794da50630e26f5de04b4c0ae5d49353f/airflow/serialization/helpers.py#L63



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Reply via email to