kyungjunleeme commented on code in PR #53357:
URL: https://github.com/apache/airflow/pull/53357#discussion_r2206398796
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -390,6 +406,48 @@ 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.start_date is None
+ or dag_run_response.data_interval_start is None
+ or dag_run_response.data_interval_end is None
+ ):
+ return None
Review Comment:
```suggestion
if not isinstance(dag_run_response, DagRunResult) or any(
field is None for field in (
dag_run_response.start_date,
dag_run_response.data_interval_start,
dag_run_response.data_interval_end,
)
):
return None
```
I think that first condition(`dag_run_response is None`) is inclued in `not
isintance(dag_run_response, DagRunResult)` and we can use `all` is better to
see. how about ur opinion?
--
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]
