Add number of queries guard in public task instances list endpoints

airflow-core/src/airflow/api_fastapi/common/db/task_instances.py

@@ -0,0 +1,40 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from sqlalchemy.orm import joinedload
+from sqlalchemy.orm.interfaces import LoaderOption
+
+from airflow.models import Base
+from airflow.models.dag_version import DagVersion
+from airflow.models.dagrun import DagRun
+from airflow.models.taskinstance import TaskInstance
+
+
+def eager_load_TI_and_TIH_for_validation(orm_model: Base | None = None) -> tuple[LoaderOption, ...]:
+    """Construct the eager loading options necessary for both TaskInstanceResponse and TaskInstanceHistoryResponse objects."""
+    if orm_model is None:
+        orm_model = TaskInstance
+
+    options: tuple[LoaderOption, ...] = (
+        joinedload(orm_model.dag_version).joinedload(DagVersion.bundle),
+        joinedload(orm_model.dag_run).options(joinedload(DagRun.dag_model)),
+    )
+    if orm_model is TaskInstance:
+        options += (joinedload(orm_model.task_instance_note),)
+    return options

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

@@ -33,6 +33,7 @@
     get_latest_version_of_dag,
 )
 from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.db.task_instances import eager_load_TI_and_TIH_for_validation
 from airflow.api_fastapi.common.parameters import (
     FilterOptionEnum,
     FilterParam,
@@ -193,8 +194,7 @@ def get_mapped_task_instances(
         select(TI)
         .where(TI.dag_id == dag_id, TI.run_id == dag_run_id, TI.task_id == task_id, TI.map_index >= 0)
         .join(TI.dag_run)
-        .options(joinedload(TI.dag_version))
-        .options(joinedload(TI.dag_run).options(joinedload(DagRun.dag_model)))
+        .options(*eager_load_TI_and_TIH_for_validation())
     )
     # 0 can mean a mapped TI that expanded to an empty list, so it is not an automatic 404
     unfiltered_total_count = get_query_count(query, session=session)
@@ -324,8 +324,7 @@ def _query(orm_object: Base) -> Select:
                 orm_object.task_id == task_id,
                 orm_object.map_index == map_index,
             )
-            .options(joinedload(orm_object.dag_version))
-            .options(joinedload(orm_object.dag_run).options(joinedload(DagRun.dag_model)))
+            .options(*eager_load_TI_and_TIH_for_validation(orm_object))
             .options(joinedload(orm_object.hitl_detail))
         )
         return query
@@ -467,11 +466,7 @@ def get_task_instances(
     """
     dag_run = None
     query = (
-        select(TI)
-        .join(TI.dag_run)
-        .outerjoin(TI.dag_version)
-        .options(joinedload(TI.dag_version))
-        .options(joinedload(TI.dag_run).options(joinedload(DagRun.dag_model)))
+        select(TI).join(TI.dag_run).outerjoin(TI.dag_version).options(*eager_load_TI_and_TIH_for_validation())
     )
     if dag_run_id != "~":
         dag_run = session.scalar(select(DagRun).filter_by(run_id=dag_run_id))
@@ -597,7 +592,9 @@ def get_task_instances_batch(
         TI,
     ).set_value([body.order_by] if body.order_by else None)
 
-    query = select(TI).join(TI.dag_run).outerjoin(TI.dag_version)
+    query = (
+        select(TI).join(TI.dag_run).outerjoin(TI.dag_version).options(*eager_load_TI_and_TIH_for_validation())
+    )
     task_instance_select, total_entries = paginated_select(
         statement=query,
         filters=[