Extra Links do not works with mapped operators #25360

from pendulum import datetime

from airflow.decorators import dag
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.operators.empty import EmptyOperator

EXTERNAL_DAG_IDS = [f"example_external_dag_{ix:02d}" for ix in range(3)]
DAG_KWARGS = {
    "start_date": datetime(2022, 7, 1),
    "schedule_interval": "@daily",
    "catchup": False,
    "tags": ["mapped_extra_links", "AIP-42", "serialization"],
}


def external_dags():
    EmptyOperator(task_id="dummy")


@dag(**DAG_KWARGS)
def external_regular_task_sensor():
    for external_dag_id in EXTERNAL_DAG_IDS:
        ExternalTaskSensor(
            task_id=f'wait_for_{external_dag_id}',
            external_dag_id=external_dag_id,
            poke_interval=5,
        )


@dag(**DAG_KWARGS)
def external_mapped_task_sensor():
    ExternalTaskSensor.partial(
        task_id='wait',
        poke_interval=5,
    ).expand(external_dag_id=EXTERNAL_DAG_IDS)


dag_external_regular_task_sensor = external_regular_task_sensor()
dag_external_mapped_task_sensor = external_mapped_task_sensor()

for dag_id in EXTERNAL_DAG_IDS:
    globals()[dag_id] = dag(dag_id=dag_id, **DAG_KWARGS)(external_dags)()