Display a more user friendly error when invalid branches are provided to branch operators #54224
Description
Body
Right now, if a branch operator doesnt have a clear branch to take, it displays:
ERROR - Task failed with exception
AirflowException: 'branch_task_ids' expected all task IDs are strings. Invalid tasks found: {(None, 'NoneType')}.
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 918 in run
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 1205 in _execute_task
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 403 in wrapper
File "/opt/airflow/providers/standard/src/airflow/providers/standard/operators/branch.py", line 102 in execute
File "/opt/airflow/providers/standard/src/airflow/providers/standard/operators/branch.py", line 44 in do_branch
File "/opt/airflow/providers/standard/src/airflow/providers/standard/utils/skipmixin.py", line 137 in skip_all_except
This can be better formatted / presented to the end users.
To repro, run this:
from datetime import timedelta
from airflow.sdk import DAG
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.python import BranchPythonOperator, PythonOperator
default_args = {
"owner": "airflow",
"depends_on_past": False,
"retries": 1,
"retry_delay": timedelta(minutes=5),
}
def predestine(**kwargs):
kwargs["ti"].xcom_push(key="choice", value="foo")
def let_fates_decide(**kwargs):
return kwargs["ti"].xcom_pull(key="choice")
with DAG(
dag_id="simple_branch",
schedule=None,
default_args=default_args,
catchup=False,
tags=["core"],
) as dag:
(
PythonOperator(task_id="predestine", python_callable=predestine)
>> BranchPythonOperator(
task_id="let_fates_decide",
python_callable=let_fates_decide,
)
>> [EmptyOperator(task_id="foo"), EmptyOperator(task_id="bar")]
)
Committer
- I acknowledge that I am a maintainer/committer of the Apache Airflow project.