Fix TriggerDagRunOperator extra_link when trigger_dag_id is templated

Author: fredthomsen

providers/src/airflow/providers/standard/operators/trigger_dagrun.py

@@ -67,11 +67,28 @@ class TriggerDagRunLink(BaseOperatorLink):
     name = "Triggered DAG"
 
     def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey) -> str:
+        from airflow.models.renderedtifields import RenderedTaskInstanceFields
+        from airflow.models.taskinstance import TaskInstance
+
+        ti = TaskInstance.get_task_instance(
+            dag_id=ti_key.dag_id, run_id=ti_key.run_id, task_id=ti_key.task_id, map_index=ti_key.map_index
+        )
+        if TYPE_CHECKING:
+            assert ti is not None
+
+        template_fields = RenderedTaskInstanceFields.get_templated_fields(ti)
+        untemplated_trigger_dag_id = cast(TriggerDagRunOperator, operator).trigger_dag_id
+        if template_fields:
+            trigger_dag_id = template_fields.get("trigger_dag_id", untemplated_trigger_dag_id)
+        else:
+            trigger_dag_id = untemplated_trigger_dag_id
+
         # Fetch the correct dag_run_id for the triggerED dag which is
         # stored in xcom during execution of the triggerING task.
         triggered_dag_run_id = XCom.get_value(ti_key=ti_key, key=XCOM_RUN_ID)
+
         query = {
-            "dag_id": cast(TriggerDagRunOperator, operator).trigger_dag_id,
+            "dag_id": trigger_dag_id,
             "dag_run_id": triggered_dag_run_id,
         }
         return build_airflow_url_with_query(query)

tests/operators/test_trigger_dagrun.py

@@ -271,6 +271,27 @@ def test_trigger_dagrun_with_templated_logical_date(self, dag_maker):
             assert triggered_dag_run.logical_date == DEFAULT_DATE
             self.assert_extra_link(triggered_dag_run, task, session)
 
+    def test_trigger_dagrun_with_templated_trigger_dag_id(self, dag_maker):
+        """Test TriggerDagRunOperator with templated trigger dag id."""
+        with dag_maker(
+            TEST_DAG_ID, default_args={"owner": "airflow", "start_date": DEFAULT_DATE}, serialized=True
+        ) as dag:
+            task = TriggerDagRunOperator(
+                task_id="__".join(["test_trigger_dagrun_with_templated_trigger_dag_id", TRIGGERED_DAG_ID]),
+                trigger_dag_id="{{ ti.task_id.rsplit('.', 1)[-1].split('__')[-1] }}",
+            )
+        self.re_sync_triggered_dag_to_db(dag, dag_maker)
+        dag_maker.create_dagrun()
+        task.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, ignore_ti_state=True)
+
+        with create_session() as session:
+            dagruns = session.query(DagRun).filter(DagRun.dag_id == TRIGGERED_DAG_ID).all()
+            assert len(dagruns) == 1
+            triggered_dag_run = dagruns[0]
+            assert triggered_dag_run.external_trigger
+            assert triggered_dag_run.dag_id == TRIGGERED_DAG_ID
+            self.assert_extra_link(triggered_dag_run, task, session)
+
     def test_trigger_dagrun_operator_conf(self, dag_maker):
         """Test passing conf to the triggered DagRun."""
         with dag_maker(