Used in https://github.com/mhenc/airflow/blob/master/airflow/dag_processing/manager.py#L613

should be straighforward
for node in group.topological_sort(_include_subdag_tasks=include_subdag_tasks):
                if isinstance(node, TaskGroup):
                    yield from nested_topo(node)
                else:
                    yield node
        return tuple(nested_topo(self.task_group))
    @provide_session
    def set_dag_runs_state(
        self,
        state: str = State.RUNNING,
        session: Session = NEW_SESSION,
        start_date: datetime | None = None,
        end_date: datetime | None = None,
        dag_ids: list[str] = [],
    ) -> None:
        warnings.warn(
            "This method is deprecated and will be removed in a future version.",
            RemovedInAirflow3Warning,
            stacklevel=3,
        )
        dag_ids = dag_ids or [self.dag_id]
        query = session.query(DagRun).filter(DagRun.dag_id.in_(dag_ids))
        if start_date:
            query = query.filter(DagRun.execution_date >= start_date)
        if end_date:
            query = query.filter(DagRun.execution_date <= end_date)
        query.update({DagRun.state: state}, synchronize_session="fetch")
    @provide_session
    def clear(