**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/settings.py
import rich.console
    return rich.console.Console(file=file)
def custom_show_warning(message, category, filename, lineno, file=None, line=None):
    """Custom function to print rich and visible warnings"""
    
    from rich.markup import escape
    msg = f"[bold]{line}" if line else f"[bold][yellow]{filename}:{lineno}"
    msg += f" {category.__name__}[/bold]: {escape(str(message))}[/yellow]"
    write_console = _get_rich_console(file or sys.stderr)
    write_console.print(msg, soft_wrap=True)
warnings.showwarning = custom_show_warning
def policy(task):  
    """
    This policy setting allows altering tasks after they are loaded in
    the DagBag. It allows administrator to rewire some task parameters.
    To define policy, add a ``airflow_local_settings`` module
    to your PYTHONPATH that defines this ``policy`` function.
    Here are a few examples of how this can be useful:
    * You could enforce a specific queue (say the ``spark`` queue)
        for tasks using the ``SparkOperator`` to make sure that these
        tasks get wired to the right workers
    * You could enforce a task timeout policy, making sure that no tasks run
        for more than 48 hours
    * ...
    """
def task_instance_mutation_hook(task_instance):