Allow to except_skip None on BranchPythonOperator

[raphaelauv]

Add a test on skip_all_except and also allow None and not only empty iterable object.

because currently this code

def skip_things() -> str:
        if NOTHING:
            return None
        elif SOMETHING_WEIRD:
            return "task_2"
        else:
            return "task_3"

branch_task = BranchPythonOperator(
    task_id='branch_task',
    python_callable=skip_things)

task_2 = ...
task_3 = ...

branch_task >> [task_2, task_3 ]

fail with :

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1165, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/usr/local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1283, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/usr/local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1313, in _execute_task
    result = task_copy.execute(context=context)
  File "/usr/local/lib/python3.8/site-packages/airflow/operators/python.py", line 180, in execute
    self.skip_all_except(context['ti'], branch)
  File "/usr/local/lib/python3.8/site-packages/airflow/models/skipmixin.py", line 128, in skip_all_except
    branch_task_ids = set(branch_task_ids)
TypeError: 'NoneType' object is not iterable

Only an empty iterable is currently allow

def skip_things() -> str:
        if NOTHING:
            return []
        elif SOMETHING_WEIRD:
            return "task_2"
        else:
            return "task_3"

branch_task = BranchPythonOperator(
    task_id='branch_task',
    python_callable=skip_things)

task_2 = ...
task_3 = ...

branch_task >> [task_2, task_3 ]

[raphaelauv]

@eladkal could you re-review this PR , than you again

[uranusjr]

I’m personally skeptical to this since it’s much too easy for a user to accidentally return None from a function and cause tasks to unintentionally not run. But I guess this is a subjective issue and not a designe is objectively superior.

[raphaelauv]

@uranusjr could you re-review pls , thank you

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[raphaelauv]

I just rebased

[eladkal]

Can you please explain the behavior of returning None in the docs? (I'm not sure if we have a designated doc specifically for this operator)

airflow/docs/apache-airflow/concepts/dags.rst

Line 317 in db76f0b

As with the callable for ``BranchPythonOperator``, this method should return the ID of a downstream task, or a list of task IDs, which will be run, and all others will be skipped::

[eladkal]

LGTM

[raphaelauv]

@eladkal can we merge ? thanks