code: def get_task_instances_batch(
      session: Session = NEW_SESSION,
  ) -> APIResponse:
      """Get list of task instances."""
      body = get_json_request_dict()
      try:
          data = task_instance_batch_form.load(body)
      except ValidationError as err:
          raise BadRequest(detail=str(err.messages))
      dag_ids = data["dag_ids"]
      if dag_ids:
          requests: Sequence[IsAuthorizedDagRequest] = [
              {
                  "method": "GET",
                  "details": DagDetails(id=id),
                  "user": g.user,
              }
              for id in dag_ids
          ]
          if not get_auth_manager().batch_is_authorized_dag(requests):
              raise PermissionDenied(
                  detail=f"User not allowed to access some of these DAGs: {list(dag_ids)}"
              )
      else:
          dag_ids = (
              get_airflow_app().appbuilder.sm.get_accessible_dag_ids(
                  g.user
              )
          )
  
      states = _convert_ti_states(data["state"])
      base_query = select(TI).join(TI.dag_run)
  
      base_query = _apply_array_filter(
          base_query, key=TI.dag_id, values=dag_ids
      )
      base_query = _apply_array_filter(
          base_query, key=TI.run_id, values=data["dag_run_ids"]
      )
      base_query = _apply_array_filter(
          base_query, key=TI.task_id, values=data["task_ids"]
      )
      base_query = _apply_range_filter(
          base_query,
          key=DR.execution_date,
          value_range=(
              data["execution_date_gte"],
              data["execution_date_lte"],
          ),
      )
      base_query = _apply_range_filter(
          base_query,
          key=TI.start_date,
          value_range=(data["start_date_gte"], data["start_date_lte"]),
      )
      base_query = _apply_range_filter(
          base_query,
          key=TI.end_date,
          value_range=(data["end_date_gte"], data["end_date_lte"]),
      )
      base_query = _apply_range_filter(
          base_query,
          key=TI.duration,
          value_range=(data["duration_gte"], data["duration_lte"]),
      )
      base_query = _apply_array_filter(
          base_query, key=TI.state, values=states
      )
      base_query = _apply_array_filter(
          base_query, key=TI.pool, values=data["pool"]
      )
      base_query = _apply_array_filter(
          base_query, key=TI.queue, values=data["queue"]
      )
  
      # Count elements before joining extra columns
      total_entries = get_query_count(base_query, session=session)
      # Add join
      base_query = base_query.join(
          SlaMiss,
          and_(
              SlaMiss.dag_id == TI.dag_id,
              SlaMiss.task_id == TI.task_id,
              SlaMiss.execution_date == DR.execution_date,
          ),
          isouter=True,
      ).add_columns(SlaMiss)
      ti_query = base_query.options(
          joinedload(TI.rendered_task_instance_fields)
      )
      # using execute because we want the SlaMiss entity. Scalars don't return None for missing entities
      task_instances = session.execute(ti_query).all()
  
      return task_instance_collection_schema.dump(
          TaskInstanceCollection(
              task_instances=task_instances, total_entries=total_entries
          )
      )
  
imports: from __future__ import annotations
  from typing import TYPE_CHECKING, Any, Iterable, Sequence, TypeVar
  from flask import g
  from marshmallow import ValidationError
  from sqlalchemy import and_, or_, select
  from sqlalchemy.exc import MultipleResultsFound
  from sqlalchemy.orm import joinedload
  from airflow.api_connexion import security
  from airflow.api_connexion.endpoints.request_dict import get_json_request_dict
  from airflow.api_connexion.exceptions import BadRequest, NotFound, PermissionDenied
  from airflow.api_connexion.parameters import format_datetime, format_parameters
  from airflow.api_connexion.schemas.task_instance_schema import TaskInstanceCollection, TaskInstanceReferenceCollection, clear_task_instance_form, set_single_task_instance_state_form, set_task_instance_note_form_schema, set_task_instance_state_form, task_instance_batch_form, task_instance_collection_schema, task_instance_reference_collection_schema, task_instance_reference_schema, task_instance_schema
  from airflow.api_connexion.security import get_readable_dags
  from airflow.auth.managers.models.resource_details import DagAccessEntity, DagDetails
  from airflow.exceptions import TaskNotFound
  from airflow.models import SlaMiss
  from airflow.models.dagrun import DagRun
  from airflow.models.operator import needs_expansion
  from airflow.models.taskinstance import TaskInstance, clear_task_instances
  from airflow.utils.airflow_flask_app import get_airflow_app
  from airflow.utils.db import get_query_count
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.state import DagRunState, TaskInstanceState
  from airflow.www.extensions.init_auth_manager import get_auth_manager
label_desc: Use image node_id 8 for calling a global function with example usage: get_task_instances_batch(session) and returns: task_instance_collection_schema