Display a more friendly error when invalid branches are provided to b…

[jroachgolf84]

Previously, an error message like the one below was output when an invalid Task ID/object containing Task ID's was returned by a python_callable as part of a BrnachPythonOperator. This message was enhanced in this PR.

Previous error message:

AirflowException: 'branch_task_ids' expected all task IDs are strings. Invalid tasks found: {(None, 'NoneType')}

closes: #54224

[jroachgolf84]

cc: @amoghrajesh

[jroachgolf84]

@amoghrajesh, I'm surprised the provider distribution tests are failing with the exception below. This PR changed this error message in both skipmixin.py and test_skipmixin.py. There is NO reference in the code base to previous error message that read something like "'branch_task_ids' expected all task IDs are strings. Invalid tasks found: {(42, 'int')}.". I'll continue to take a look, but I'm curious if you have any thoughts?

providers/standard/tests/unit/standard/utils/test_skipmixin.py::TestSkipMixin::test_raise_exception_on_not_accepted_iterable_branch_task_ids_type - AssertionError: Regex pattern did not match.
 Regex: "Cannot properly branch. Invalid branch task ID's were returned by the python_callable. These invalid 'branch_task_ids' were: \\{\\(42, 'int'\\)\\}\\. Make sure your python_callable returns an Iterable of strings storing valid task_id's. These task_id's should correspond to Tasks within your DAG."
 Input: "'branch_task_ids' expected all task IDs are strings. Invalid tasks found: {(42, 'int')}."

[amoghrajesh]

@jroachgolf84 it's failing the compat tests because in 2.11 skipmixin was part of core: airflow/models/skipmixin.py and if you observe the stack trace, it says:

providers/standard/tests/unit/standard/utils/test_skipmixin.py:337: in test_raise_exception_on_not_accepted_iterable_branch_task_ids_type
    SkipMixin().skip_all_except(ti=ti1, branch_task_ids=["task", 42])
/usr/local/lib/python3.10/site-packages/airflow/models/skipmixin.py:192: in skip_all_except
    SkipMixin._skip_all_except(ti=ti, branch_task_ids=branch_task_ids)
/usr/local/lib/python3.10/site-packages/airflow/api_internal/internal_api_call.py:166: in wrapper
    return func(*args, **kwargs)
/usr/local/lib/python3.10/site-packages/airflow/utils/session.py:97: in wrapper
    return func(*args, session=session, **kwargs)
/usr/local/lib/python3.10/site-packages/airflow/models/skipmixin.py:221: in _skip_all_except
    raise AirflowException(
E   airflow.exceptions.AirflowException: 'branch_task_ids' expected all task IDs are strings. Invalid tasks found: {(42, 'int')}.

It is going into models/skipmixin and raising the exception from there (original 'branch_task_ids' expected all task IDs are strings.....) and hence it is not matching what you expect it to match.

Unfortunately there isn't a way to fix in 2.x and it is not important enough to fix too, so I would suggest you validate it by version, like so:

    if AIRFLOW_V_3_0_PLUS:
        # New improved error message for Airflow 3.0+
        error_message = (
            r"Cannot properly branch\. Invalid branch task ID's were returned by the python_callable\. "
            r"These invalid 'branch_task_ids' were: \{\(42, 'int'\)\}\. "
            r"Make sure your python_callable returns an Iterable of strings storing valid task_id's\. "
            r"These task_id's should correspond to Tasks within your DAG\."
        )
    else:
        # Old error message for Airflow 2.x
        error_message = (
            r"'branch_task_ids' expected all task IDs are strings\. "
            r"Invalid tasks found: \{\(42, 'int'\)\}\."
        )

[jroachgolf84]

@amoghrajesh, thanks for the tip. The tests have been updated appropriately!

[amoghrajesh]

@jroachgolf84 I think we can make it more user friendly

[jroachgolf84]

@amoghrajesh, mind taking a peek?