dag_run.get_task_instances alternative in airflow 3

[rodigu]

couldn't find an alternative to the dag_run.get_task_instances() function for airflow 3.

i need a list of the task instances in the current dag run, as well as their respective statuses.

is there a new way to access this information through context?

[gopidesupavan]

use this way to get the task states. this methods supports following args:

    dag_id,
    map_index,
    task_ids,
    task_group_id,
    logical_dates,
    run_ids

from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
RuntimeTaskInstance.get_task_states()

[nehemias-bedrockanalytics]

Awesome, thanks. How could we get the complete task instances objects ? not just the states, I want to read values like map_index, start_date, try_number, max_tries, etc.

[nehemias-bedrockanalytics]

Not sure what is the right way but looking at RuntimeTaskInstance implementation I see how communication with the Supervisor is made though the SUPERVISOR_COMMS object, which seems to have a list of things it can do, of which GetTaskStates is one of them.

comms.py has a list of the available actions we can take, I don't see one to get task instances, I see another useful one to me, TriggerDagRun.

This would be a way to trigger a dag run:

from datetime import datetime, timezone
from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
from airflow.sdk.execution_time.comms import TriggerDagRun

trigger = TriggerDagRun(dag_id="my_dag", 
                  run_id="my_dag_run_001", 
                  logical_date=datetime.now(timezone.utc),
                  conf={})

response = SUPERVISOR_COMMS.send(msg=trigger)

response will be something like

OKResponse(ok=True, type='OKResponse')

Still want to know how to get task instances properly in Airflow 3

[karenbraganz]

@nehemias-bedrockanalytics you could use the Airflow REST API to get task instances without DB access. This endpoint of the API returns task instances if you provide a dag_id and run_id.

[nehemias-bedrockanalytics]

@nehemias-bedrockanalytics you could use the Airflow REST API to get task instances without DB access. This endpoint of the API returns task instances if you provide a dag_id and run_id.

Thank you for this suggestion, I am trying to avoid the REST API as much as I can just because I would need to provide user credentials from within the Dag Run, which seems odd given we are already inside Airflow, an approach like SUPERVISOR_COMMS would be ideal. Or perhaps there's a way to authenticate to the REST API from within a Dag Run ?