yuqian90 commented on a change in pull request #7276: [AIRFLOW-5391] Do not run
skipped tasks when they are cleared
URL: https://github.com/apache/airflow/pull/7276#discussion_r379379912
##########
File path: airflow/operators/branch_operator.py
##########
@@ -48,4 +48,21 @@ def choose_branch(self, context: Dict) -> Union[str,
Iterable[str]]:
raise NotImplementedError
def execute(self, context: Dict):
- self.skip_all_except(context['ti'], self.choose_branch(context))
+ branch = self.choose_branch(context)
+ self.skip_all_except(context['ti'], branch)
+ return branch
+
+ def evaluate_skip_condition(self, xcom_value: Union[str, List[str]],
downstream_ti: TaskInstance):
+ """
+ :param xcom_value: The task_id of list of task_id returned by
execute().
+ :type xcom_value: str or list[str]
+ :param downstream_ti: The TaskInstance of child task
+ :type downstream_ti: airflow.models.TaskInstance
+ :return: True if xcom_value indicates downstream_ti should be skipped.
Otherwise False.
+ :rtype: bool
+ """
+ branch = xcom_value
+ if isinstance(branch, str):
+ branch = [branch]
+
+ return downstream_ti.task_id not in branch
Review comment:
@ashb how about this:
Since there's `skip()` and `skip_all_except()`. I can make `skip()` store
what's skipped, and make `skip_all_except()` store what's in the "except".
So the XCom data will look something along this line:
```
"skipped": [B, C]
"except": [A]
```
When looking up this XCom values, first, skip [B, C], and then also skip
everything that's not equal in "except".
That should cover all the edge cases that can happen.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
