code: def post_clear_task_instances(
      *, dag_id: str, session: Session = NEW_SESSION
  ) -> APIResponse:
      """Clear task instances."""
      body = get_json_request_dict()
      try:
          data = clear_task_instance_form.load(body)
      except ValidationError as err:
          raise BadRequest(detail=str(err.messages))
  
      dag = get_airflow_app().dag_bag.get_dag(dag_id)
      if not dag:
          error_message = f"Dag id {dag_id} not found"
          raise NotFound(error_message)
      reset_dag_runs = data.pop("reset_dag_runs")
      dry_run = data.pop("dry_run")
      # We always pass dry_run here, otherwise this would try to confirm on the terminal!
      dag_run_id = data.pop("dag_run_id", None)
      future = data.pop("include_future", False)
      past = data.pop("include_past", False)
      downstream = data.pop("include_downstream", False)
      upstream = data.pop("include_upstream", False)
      if dag_run_id is not None:
          dag_run: DR | None = session.scalar(
              select(DR).where(
                  DR.dag_id == dag_id, DR.run_id == dag_run_id
              )
          )
          if dag_run is None:
              error_message = (
                  f"Dag Run id {dag_run_id} not found in dag {dag_id}"
              )
              raise NotFound(error_message)
          data["start_date"] = dag_run.logical_date
          data["end_date"] = dag_run.logical_date
      if past:
          data["start_date"] = None
      if future:
          data["end_date"] = None
      task_ids = data.pop("task_ids", None)
      if task_ids is not None:
          task_id = [
              task[0] if isinstance(task, tuple) else task
              for task in task_ids
          ]
          dag = dag.partial_subset(
              task_ids_or_regex=task_id,
              include_downstream=downstream,
              include_upstream=upstream,
          )
  
          if len(dag.task_dict) > 1:
              # If we had upstream/downstream etc then also include those!
              task_ids.extend(
                  tid for tid in dag.task_dict if tid != task_id
              )
      task_instances = dag.clear(
          dry_run=True,
          dag_bag=get_airflow_app().dag_bag,
          task_ids=task_ids,
          **data,
      )
  
      if not dry_run:
          clear_task_instances(
              task_instances,
              session,
              dag=dag,
              dag_run_state=DagRunState.QUEUED
              if reset_dag_runs
              else False,
          )
  
      return task_instance_reference_collection_schema.dump(
          TaskInstanceReferenceCollection(task_instances=task_instances)
      )
  
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 9 for calling a global function with example usage: post_clear_task_instances() and returns: task_instance_reference_collection_schema