Option for DAG to throw exception when dependency between tasks are defined more than once

[notatallshaw-gts]

Currently if you set the same dependency more than once Airflow silently accepts it (except for airflow tasks test which gives a vague message), e.g.

from datetime import datetime

from airflow.models import DAG
from airflow.operators.empty import EmptyOperator

with DAG(
    dag_id="test", schedule_interval="@once", start_date=datetime.min
) as dag:
    task_1 = EmptyOperator(task_id="one")
    task_2 = EmptyOperator(task_id="two")

    task_1 >> task_2
    task_1 >> task_2

This at best indicates a code smell and at worst indicates a logic error, and while it's simple to spot in this DAG it can be difficult to spot in highly complex DAGs. It would be great to have a DAG parameter like strict_task_dependency which when set to True would throw an exception in this situation.

Apologies if this has already been discussed and I missed it.

[potiuk]

Feel Free to open PR with such check.

[notatallshaw-gts]

I'll give it a go!

[potiuk]

Feel free. Just open PR

[notatallshaw]

Crested a first attempt at a PR: #30189

[notatallshaw-gts]

Answer is no, this use case is too narrow: #30189

I think though there is room for a general use case of the user being able to add a callback to DAG construction events to enforce DAG quality, then this and many other use cases could be implemented by users without needing to add specialized logic in Airflow for each possible quality check.

But I don't see myself making that PR soon, so marking as answered for now.