code: def get_mapped_task_instances(
      *,
      dag_id: str,
      dag_run_id: str,
      task_id: str,
      execution_date_gte: str | None = None,
      execution_date_lte: str | None = None,
      start_date_gte: str | None = None,
      start_date_lte: str | None = None,
      end_date_gte: str | None = None,
      end_date_lte: str | None = None,
      updated_at_gte: str | None = None,
      updated_at_lte: str | None = None,
      duration_gte: float | None = None,
      duration_lte: float | None = None,
      state: list[str] | None = None,
      pool: list[str] | None = None,
      queue: list[str] | None = None,
      limit: int | None = None,
      offset: int | None = None,
      order_by: str | None = None,
      session: Session = NEW_SESSION,
  ) -> APIResponse:
      """Get list of task instances."""
      # Because state can be 'none'
      states = _convert_ti_states(state)
  
      base_query = (
          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)
      )
  
      # 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(
          base_query, session=session
      )
      if unfiltered_total_count == 0:
          dag = get_airflow_app().dag_bag.get_dag(dag_id)
          if not dag:
              error_message = f"DAG {dag_id} not found"
              raise NotFound(error_message)
          try:
              task = dag.get_task(task_id)
          except TaskNotFound:
              error_message = f"Task id {task_id} not found"
              raise NotFound(error_message)
          if not needs_expansion(task):
              error_message = f"Task id {task_id} is not mapped"
              raise NotFound(error_message)
  
      # Other search criteria
      base_query = _apply_range_filter(
          base_query,
          key=DR.execution_date,
          value_range=(execution_date_gte, execution_date_lte),
      )
      base_query = _apply_range_filter(
          base_query,
          key=TI.start_date,
          value_range=(start_date_gte, start_date_lte),
      )
      base_query = _apply_range_filter(
          base_query,
          key=TI.end_date,
          value_range=(end_date_gte, end_date_lte),
      )
      base_query = _apply_range_filter(
          base_query,
          key=TI.duration,
          value_range=(duration_gte, duration_lte),
      )
      base_query = _apply_range_filter(
          base_query,
          key=TI.updated_at,
          value_range=(updated_at_gte, updated_at_lte),
      )
      base_query = _apply_array_filter(
          base_query, key=TI.state, values=states
      )
      base_query = _apply_array_filter(
          base_query, key=TI.pool, values=pool
      )
      base_query = _apply_array_filter(
          base_query, key=TI.queue, values=queue
      )
  
      # Count elements before joining extra columns
      total_entries = get_query_count(base_query, session=session)
  
      # Add SLA miss
      entry_query = (
          base_query.outerjoin(
              SlaMiss,
              and_(
                  SlaMiss.dag_id == TI.dag_id,
                  SlaMiss.task_id == TI.task_id,
                  SlaMiss.execution_date == DR.execution_date,
              ),
          )
          .add_columns(SlaMiss)
          .options(joinedload(TI.rendered_task_instance_fields))
      )
  
      if order_by:
          if order_by == "state":
              entry_query = entry_query.order_by(
                  TI.state.asc(), TI.map_index.asc()
              )
          elif order_by == "-state":
              entry_query = entry_query.order_by(
                  TI.state.desc(), TI.map_index.asc()
              )
          elif order_by == "-map_index":
              entry_query = entry_query.order_by(TI.map_index.desc())
          else:
              raise BadRequest(
                  detail=f"Ordering with '{order_by}' is not supported"
              )
      else:
          entry_query = entry_query.order_by(TI.map_index.asc())
  
      # using execute because we want the SlaMiss entity. Scalars don't return None for missing entities
      task_instances = session.execute(
          entry_query.offset(offset).limit(limit)
      ).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 3 for calling a global function with example usage: get_mapped_task_instances() and returns: task_instance_collection_schema