https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
stacklevel=2,
        )
        if not activate_dag_runs:
            dag_run_state = False
    if dag_run_state is not False and tis:
        from airflow.models.dagrun import DagRun  
        run_ids_by_dag_id = defaultdict(set)
        for instance in tis:
            run_ids_by_dag_id[instance.dag_id].add(instance.run_id)
        drs = (
            session.query(DagRun)
            .filter(
                or_(
                    and_(DagRun.dag_id == dag_id, DagRun.run_id.in_(run_ids))
                    for dag_id, run_ids in run_ids_by_dag_id.items()
                )
            )
            .all()
        )
        dag_run_state = DagRunState(dag_run_state)  
        for dr in drs:
            if dr.state in State.finished_dr_states:
                dr.state = dag_run_state
                dr.start_date = timezone.utcnow()
                if dag_run_state == DagRunState.QUEUED:
                    dr.last_scheduling_decision = None
                    dr.start_date = None
                    dr.clear_number += 1
    session.flush()
def _is_mappable_value(value: Any) -> TypeGuard[Collection]: