kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4324016733
##########
airflow-core/src/airflow/models/task_coordinates.py:
##########
@@ -839,3 +926,18 @@ def select_producer_ids(
region_index=request.region_index,
session=self.session,
)
+
+
+def get_task_instance_execution_position(
+ task_instance: TaskInstance, *, session: Session
+) -> TaskInstanceExecutionPosition:
+ """
+ Place a stored task instance in its mapped expansion and loop pass.
+
+ This is what the Execution API sends a worker when the task starts
(``TIRunContext.loop`` and
+ ``TIRunContext.expanded_ti_count``), so server-side consumers such as
listeners reporting a manual state
+ change describe the task instance as the worker would, without reading
region rows themselves.
+ """
+ return TaskCoordinateResolver(DBDagBag(load_op_links=False),
session).get_execution_position(
Review Comment:
One follow-up, since that helper existed for a reason. The OpenLineage
provider needs the same loop and expansion data for events it emits from the
API server or scheduler: manual state changes and executor-reported failures
(_on_task_instance_manual_state_change). Those TIs have no TIRunContext. Today
OL reads dynamic_region rows itself to get the iteration, which ties it to the
table layout. The helper was meant to give it a supported call instead,
returning the same data ti_run sends to the worker. With it gone, OL would do:
```
resolver = TaskCoordinateResolver.for_dag(ti.task.dag, session)
loop = resolver.get_loop_position(ti)
expanded_ti_count = resolver.get_expanded_ti_count(ti)
```
That works, but TaskCoordinateResolver is a core internal in airflow.models.
Are you OK with providers relying on it for this (keeping for_dag,
get_loop_position and get_expanded_ti_count stable)? Or would you prefer a
different entry point, e.g. a TaskInstance method that takes the session?
--
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]
