[API] Use BulkTaskInstanceBody for patching tis with new state (apache#57226)

Author: Prajwal7842

airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py

@@ -923,11 +923,23 @@ def patch_task_instance(
 
     for key, _ in data.items():
         if key == "new_state":
+            # Create BulkTaskInstanceBody object with map_index field
+            bulk_ti_body = BulkTaskInstanceBody(
+                task_id=task_id,
+                map_index=map_index,
+                new_state=body.new_state,
+                note=body.note,
+                include_upstream=body.include_upstream,
+                include_downstream=body.include_downstream,
+                include_future=body.include_future,
+                include_past=body.include_past,
+            )
+
             _patch_task_instance_state(
                 task_id=task_id,
                 dag_run_id=dag_run_id,
                 dag=dag,
-                task_instance_body=body,
+                task_instance_body=bulk_ti_body,
                 data=data,
                 session=session,
             )

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py

@@ -3904,8 +3904,16 @@ def test_should_update_mapped_task_instance_state(self, test_client, session):
         ti = TaskInstance(
             task=tis[0].task, run_id=tis[0].run_id, map_index=map_index, dag_version_id=tis[0].dag_version_id
         )
+        ti_2 = TaskInstance(
+            task=tis[0].task,
+            run_id=tis[0].run_id,
+            map_index=map_index + 1,
+            dag_version_id=tis[0].dag_version_id,
+        )
         ti.rendered_task_instance_fields = RTIF(ti, render_templates=False)
+        ti_2.rendered_task_instance_fields = RTIF(ti_2, render_templates=False)
         session.add(ti)
+        session.add(ti_2)
         session.commit()
 
         response = test_client.patch(
@@ -3920,6 +3928,11 @@ def test_should_update_mapped_task_instance_state(self, test_client, session):
         assert response2.status_code == 200
         assert response2.json()["state"] == self.NEW_STATE
 
+        response3 = test_client.get(f"{self.ENDPOINT_URL}/{map_index + 1}")
+        assert response3.status_code == 200
+        assert response3.json()["state"] != self.NEW_STATE
+        assert response3.json()["state"] is None
+
     def test_should_update_mapped_task_instance_summary_state(self, test_client, session):
         tis = self.create_task_instances(session)