refactor: Removed to_runtime method on TaskInstance

Author: dabla

airflow-core/src/airflow/models/taskinstance.py

@@ -1430,35 +1430,10 @@ def update_heartbeat(self):
                 .values(last_heartbeat_at=timezone.utcnow())
             )
 
-    def to_runtime(self) -> RuntimeTaskInstance:
-        from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
-
-        return RuntimeTaskInstance.model_construct(
-            id=self.id,
-            task_id=self.task_id,
-            dag_id=self.dag_id,
-            run_id=self.run_id,
-            try_number=self.try_number,
-            dag_version_id=self.dag_version_id,
-            map_index=self.map_index,
-            hostname=self.hostname,
-            task=self.task,
-            max_tries=self.max_tries,
-            start_date=self.start_date,
-            end_date=self.end_date,
-            state=self.state,
-            rendered_map_index=self.rendered_map_index,
-        )
-
     @property
     def start_trigger_args(self) -> StartTriggerArgs | None:
-        if self.task:
-            if self.task.is_mapped:
-                context = self.to_runtime().get_template_context()
-                if self.task.expand_start_from_trigger(context=context):
-                    return self.task.expand_start_trigger_args(context=context)
-            elif self.task.start_from_trigger is True:
-                return self.task.start_trigger_args
+        if self.task and self.task.start_from_trigger is True:
+            return self.task.start_trigger_args
         return None
 
     # TODO: We have some code duplication here and in the _create_ti_state_update_query_and_update_state

airflow-core/tests/unit/models/test_taskinstance.py

@@ -2583,30 +2583,6 @@ def test_task_instance_history_with_hitl_history_is_created_when_ti_goes_for_ret
         assert len(hitl_histories) == 1
         assert str(hitl_histories[0].task_instance.id) == try_id
 
-    def test_to_runtime(self, create_task_instance):
-        ti = create_task_instance(dag_id="test_to_runtime")
-        runtime_ti = ti.to_runtime()
-
-        assert ti.id == runtime_ti.id
-        assert ti.task_id == runtime_ti.task_id
-        assert ti.dag_id == runtime_ti.dag_id
-        assert ti.run_id == runtime_ti.run_id
-        assert ti.try_number == runtime_ti.try_number
-        assert ti.dag_version_id == runtime_ti.dag_version_id
-        assert ti.map_index == runtime_ti.map_index
-        assert ti.hostname == runtime_ti.hostname
-        assert ti.task == runtime_ti.task
-        assert not runtime_ti.bundle_instance
-        assert not runtime_ti._cached_template_context
-        assert not runtime_ti._ti_context_from_server
-        assert ti.max_tries == runtime_ti.max_tries
-        assert ti.start_date == runtime_ti.start_date
-        assert ti.end_date == runtime_ti.end_date
-        assert ti.state == runtime_ti.state
-        assert not runtime_ti.is_mapped
-        assert ti.rendered_map_index == runtime_ti.rendered_map_index
-        assert not runtime_ti.sentry_integration
-
 
 @pytest.mark.parametrize("pool_override", [None, "test_pool2"])
 @pytest.mark.parametrize("queue_by_policy", [None, "forced_queue"])