amoghrajesh commented on code in PR #65590:
URL: https://github.com/apache/airflow/pull/65590#discussion_r3117244471
##########
airflow-core/src/airflow/triggers/base.py:
##########
@@ -192,6 +192,51 @@ async def cleanup(self) -> None:
and handle it appropriately (in async-compatible way).
"""
+ async def get_task_state(self) -> TaskInstanceState | None:
+ """
+ Query the current state of the task instance via the Execution API.
+
+ Returns ``None`` if no task instance is associated or if the state
+ cannot be determined.
+ """
+ if self._task_instance is None:
+ return None
+ from asgiref.sync import sync_to_async
+
+ from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
+
+ ti = self._task_instance
+ task_states = await sync_to_async(RuntimeTaskInstance.get_task_states)(
+ dag_id=ti.dag_id,
+ task_ids=[ti.task_id],
+ run_ids=[ti.run_id],
+ map_index=ti.map_index,
+ )
+ try:
+ return task_states[ti.run_id][ti.task_id]
+ except (KeyError, TypeError):
+ return None
+
+ async def safe_to_cancel(self) -> bool:
Review Comment:
Outdated now.
##########
airflow-core/src/airflow/triggers/base.py:
##########
@@ -192,6 +192,51 @@ async def cleanup(self) -> None:
and handle it appropriately (in async-compatible way).
"""
+ async def get_task_state(self) -> TaskInstanceState | None:
Review Comment:
Handled this as suggested now.
--
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]
