code: def set_state(
      *,
      tasks: Collection[Operator | tuple[Operator, int]],
      run_id: str | None = None,
      execution_date: datetime | None = None,
      upstream: bool = False,
      downstream: bool = False,
      future: bool = False,
      past: bool = False,
      state: TaskInstanceState = TaskInstanceState.SUCCESS,
      commit: bool = False,
      session: SASession = NEW_SESSION,
  ) -> list[TaskInstance]:
      """
      Set the state of a task instance and if needed its relatives.
  
      Can set state for future tasks (calculated from run_id) and retroactively
      for past tasks. Will verify integrity of past dag runs in order to create
      tasks that did not exist. It will not create dag runs that are missing
      on the schedule (but it will, as for subdag, dag runs if needed).
  
      :param tasks: the iterable of tasks or (task, map_index) tuples from which to work.
          ``task.dag`` needs to be set
      :param run_id: the run_id of the dagrun to start looking from
      :param execution_date: the execution date from which to start looking (deprecated)
      :param upstream: Mark all parents (upstream tasks)
      :param downstream: Mark all siblings (downstream tasks) of task_id, including SubDags
      :param future: Mark all future tasks on the interval of the dag up until
          last execution date.
      :param past: Retroactively mark all tasks starting from start_date of the DAG
      :param state: State to which the tasks need to be set
      :param commit: Commit tasks to be altered to the database
      :param session: database session
      :return: list of tasks that have been created and updated
      """
      if not tasks:
          return []
  
      if not exactly_one(execution_date, run_id):
          raise ValueError(
              "Exactly one of dag_run_id and execution_date must be set"
          )
  
      if execution_date and not timezone.is_localized(execution_date):
          raise ValueError(
              f"Received non-localized date {execution_date}"
          )
  
      task_dags = {
          task[0].dag if isinstance(task, tuple) else task.dag
          for task in tasks
      }
      if len(task_dags) > 1:
          raise ValueError(
              f"Received tasks from multiple DAGs: {task_dags}"
          )
      dag = next(iter(task_dags))
      if dag is None:
          raise ValueError("Received tasks with no DAG")
  
      if execution_date:
          run_id = dag.get_dagrun(
              execution_date=execution_date, session=session
          ).run_id
      if not run_id:
          raise ValueError("Received tasks with no run_id")
  
      dag_run_ids = get_run_ids(
          dag, run_id, future, past, session=session
      )
      task_id_map_index_list = list(
          find_task_relatives(tasks, downstream, upstream)
      )
      task_ids = [
          task_id if isinstance(task_id, str) else task_id[0]
          for task_id in task_id_map_index_list
      ]
  
      confirmed_infos = list(
          _iter_existing_dag_run_infos(
              dag, dag_run_ids, session=session
          )
      )
      confirmed_dates = [info.logical_date for info in confirmed_infos]
  
      sub_dag_run_ids = list(
          _iter_subdag_run_ids(
              dag,
              session,
              DagRunState(state),
              task_ids,
              commit,
              confirmed_infos,
          ),
      )
  
      # now look for the task instances that are affected
  
      qry_dag = get_all_dag_task_query(
          dag, session, state, task_id_map_index_list, dag_run_ids
      )
  
      if commit:
          tis_altered = session.scalars(qry_dag.with_for_update()).all()
          if sub_dag_run_ids:
              qry_sub_dag = all_subdag_tasks_query(
                  sub_dag_run_ids, session, state, confirmed_dates
              )
              tis_altered += session.scalars(
                  qry_sub_dag.with_for_update()
              ).all()
          for task_instance in tis_altered:
              # The try_number was decremented when setting to up_for_reschedule and deferred.
              # Increment it back when changing the state again
              if task_instance.state in (
                  TaskInstanceState.DEFERRED,
                  TaskInstanceState.UP_FOR_RESCHEDULE,
              ):
                  task_instance._try_number += 1
              task_instance.set_state(state, session=session)
          session.flush()
      else:
          tis_altered = session.scalars(qry_dag).all()
          if sub_dag_run_ids:
              qry_sub_dag = all_subdag_tasks_query(
                  sub_dag_run_ids, session, state, confirmed_dates
              )
              tis_altered += session.scalars(qry_sub_dag).all()
      return tis_altered
  
imports: from __future__ import annotations
  from typing import TYPE_CHECKING, Collection, Iterable, Iterator, NamedTuple
  from sqlalchemy import or_, select
  from sqlalchemy.orm import lazyload
  from airflow.models.dagrun import DagRun
  from airflow.models.taskinstance import TaskInstance
  from airflow.operators.subdag import SubDagOperator
  from airflow.utils import timezone
  from airflow.utils.helpers import exactly_one
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.state import DagRunState, State, TaskInstanceState
  from airflow.utils.types import DagRunType
label_desc: Use image node_id 2 for calling a global function with example usage: set_state() and returns: tis_altered, list