code: def post_set_task_instances_state(
      *, dag_id: str, session: Session = NEW_SESSION
  ) -> APIResponse:
      """Set a state of task instances."""
      body = get_json_request_dict()
      try:
          data = set_task_instance_state_form.load(body)
      except ValidationError as err:
          raise BadRequest(detail=str(err.messages))
  
      error_message = f"Dag ID {dag_id} not found"
      dag = get_airflow_app().dag_bag.get_dag(dag_id)
      if not dag:
          raise NotFound(error_message)
  
      task_id = data["task_id"]
      task = dag.task_dict.get(task_id)
  
      if not task:
          error_message = f"Task ID {task_id} not found"
          raise NotFound(error_message)
  
      execution_date = data.get("execution_date")
      run_id = data.get("dag_run_id")
      if (
          execution_date
          and (
              session.scalars(
                  select(TI).where(
                      TI.task_id == task_id,
                      TI.dag_id == dag_id,
                      TI.execution_date == execution_date,
                  )
              ).one_or_none()
          )
          is None
      ):
          raise NotFound(
              detail=f"Task instance not found for task {task_id!r} on execution_date {execution_date}"
          )
  
      if run_id and not session.get(
          TI,
          {
              "task_id": task_id,
              "dag_id": dag_id,
              "run_id": run_id,
              "map_index": -1,
          },
      ):
          error_message = f"Task instance not found for task {task_id!r} on DAG run with ID {run_id!r}"
          raise NotFound(detail=error_message)
  
      tis = dag.set_task_instance_state(
          task_id=task_id,
          run_id=run_id,
          execution_date=execution_date,
          state=data["new_state"],
          upstream=data["include_upstream"],
          downstream=data["include_downstream"],
          future=data["include_future"],
          past=data["include_past"],
          commit=not data["dry_run"],
          session=session,
      )
      return task_instance_reference_collection_schema.dump(
          TaskInstanceReferenceCollection(task_instances=tis)
      )
  
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 10 for calling a global function with example usage: post_set_task_instances_state() and returns: task_instance_reference_collection_schema