uranusjr commented on code in PR #53357:
URL: https://github.com/apache/airflow/pull/53357#discussion_r2238296864
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -388,6 +403,49 @@ def source_task_instance(self) ->
AssetEventSourceTaskInstance | None:
map_index=self.source_map_index,
)
+ @cached_property
+ def source_dag_run(self) -> AssetEventSourceDagRun | None:
+ if not (self.source_dag_id and self.source_run_id):
+ return None
+ # Use the supervisor to get the DagRun information via API
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ try:
+ # Send request to supervisor to get DagRun information
+ dag_run_response = SUPERVISOR_COMMS.send(
+ GetDagRun(dag_id=self.source_dag_id, run_id=self.source_run_id)
+ )
+
+ if dag_run_response is None:
+ return None
+
+ # Type guard to ensure we have a DagRunResult
+ if not isinstance(dag_run_response, DagRunResult):
+ return None
+
+ # Check for required fields and provide defaults for optional
datetime fields
+ if (
+ dag_run_response.logical_date is None
+ or dag_run_response.start_date is None
+ or dag_run_response.data_interval_start is None
+ or dag_run_response.data_interval_end is None
+ ):
+ return None
+
+ return AssetEventSourceDagRun(
+ dag_id=dag_run_response.dag_id,
+ run_id=dag_run_response.run_id,
+ logical_date=dag_run_response.logical_date,
+ start_date=dag_run_response.start_date,
+ end_date=dag_run_response.end_date,
+ state=dag_run_response.state,
+ data_interval_start=dag_run_response.data_interval_start,
+ data_interval_end=dag_run_response.data_interval_end,
+ )
Review Comment:
Why do we need to build this instead of just exposing `DagRunResponse`?
--
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]
