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

should be straighforward
.group_by(DagModel.dag_id)
                    .having(func.count() >= func.max(DagModel.max_active_runs))
                    .all()
                )
            }
            if exclusion_list:
                dataset_triggered_dag_ids -= exclusion_list
                dataset_triggered_dag_info = {
                    k: v for k, v in dataset_triggered_dag_info.items() if k not in exclusion_list
                }
        
        query = (
            session.query(cls)
            .filter(
                cls.is_paused == expression.false(),
                cls.is_active == expression.true(),
                cls.has_import_errors == expression.false(),
                or_(
                    cls.next_dagrun_create_after <= func.now(),
                    cls.dag_id.in_(dataset_triggered_dag_ids),
                ),
            )
            .order_by(cls.next_dagrun_create_after)
            .limit(cls.NUM_DAGS_PER_DAGRUN_QUERY)
        )
        return (
            with_row_locks(query, of=cls, session=session, **skip_locked(session=session)),
            dataset_triggered_dag_info,
        )
    def calculate_dagrun_date_fields(