amoghrajesh commented on code in PR #44843:
URL: https://github.com/apache/airflow/pull/44843#discussion_r1883375971
##########
airflow/serialization/helpers.py:
##########
@@ -43,6 +43,10 @@ def is_jsonable(x):
max_length = conf.getint("core", "max_templated_field_length")
+ # Handle empty set or tuple explicitly
+ if isinstance(template_field, (set, tuple)) and not template_field:
+ return []
Review Comment:
Yeah i was thinking of something similar. This looks fine for now
##########
airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -237,7 +237,7 @@ def ti_heartbeat(
)
def ti_put_rtif(
task_instance_id: UUID,
- put_rtif_payload: RTIFPayload,
Review Comment:
ACK. Sounds good
##########
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:
Right, instead of duplication at the moment, we can reuse it and when we
port the logic for templating, we anyways are going to revisit this. Better to
handle it then. We will not miss it too cos we have a TODO
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
