amoghrajesh commented on code in PR #44725:
URL: https://github.com/apache/airflow/pull/44725#discussion_r1873297988
##########
task_sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -136,11 +143,27 @@ def startup() -> tuple[RuntimeTaskInstance, Logger]:
# TODO: set the "magic loop" context vars for parsing
ti = parse(msg)
log.debug("DAG file parsed", file=msg.file)
- return ti, log
else:
raise RuntimeError(f"Unhandled startup message {type(msg)} {msg}")
# TODO: Render fields here
+ # 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.
+
+ # 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=RTIFPayload(**payload))
Review Comment:
Might have to do a slightly better job here, this could break if we end up
calling for any normal task. We do not want to send these requests until we are
fully functional. Unit tests pass here.
--
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]
