**Description**

Currently we have **only** task level [cluster policy](https://airflow.readthedocs.io/en/latest/concepts.html?highlight=cluster%20policy#cluster-policy):
https://github.com/apache/airflow/blob/7dfb3980cecfafdb7d6b79d17455b08971cec7d4/airflow/models/dagbag.py#L386-L387

This is an amazing tool that allows users to validate their tasks by either skipping DAG or setting default values (queues, owners, emails etc). I would like to proposes same mechanism for DAGs.

Some may argue that users already can implement it by accessing `task.dag` attribute in the policy, however this has the following drawbacks:
- it's not "explicit over implicit" see https://github.com/apache/airflow/issues/12086#issuecomment-722986766
- the "DAG policy" is executed for every task (if there's 1000 tasks in DAG, the DAG policy will be checked 1000 times)

 **Use case / motivation**

Create explicit mechanism for creating "DAG level" cluster policy that will be optimised for this task (check only once per dag not for every task) and will not required users to implement this custom optimisation logic (for example by monkey patching or changes in their fork).

**Related Issues**

#10282

airflow/models/dagbag.py
)
                if sd_last_updated_datetime > self.dags_last_fetched[dag_id]:
                    self._add_dag_from_db(dag_id=dag_id, session=session)
            return self.dags.get(dag_id)
        
        dag = None
        root_dag_id = dag_id
        if dag_id in self.dags:
            dag = self.dags[dag_id]
            if dag.is_subdag:
                root_dag_id = dag.parent_dag.dag_id  
        
        orm_dag = DagModel.get_current(root_dag_id, session=session)
        if not orm_dag:
            return self.dags.get(dag_id)
        
        is_missing = root_dag_id not in self.dags
        is_expired = orm_dag.last_expired and dag and dag.last_loaded < orm_dag.last_expired
        if is_missing or is_expired:
            
            found_dags = self.process_file(
                filepath=correct_maybe_zipped(orm_dag.fileloc), only_if_updated=False
            )
            
            if found_dags and dag_id in [found_dag.dag_id for found_dag in found_dags]:
                return self.dags[dag_id]
            elif dag_id in self.dags:
                del self.dags[dag_id]
        return self.dags.get(dag_id)
    def _add_dag_from_db(self, dag_id: str, session: Session):