shahar1 commented on code in PR #46584:
URL: https://github.com/apache/airflow/pull/46584#discussion_r1947935272
##########
providers/standard/src/airflow/providers/standard/operators/python.py:
##########
@@ -272,48 +274,25 @@ def __init__(self, *, ignore_downstream_trigger_rules:
bool = True, **kwargs) ->
def execute(self, context: Context) -> Any:
condition = super().execute(context)
- self.log.info("Condition result is %s", condition)
-
- if condition:
- self.log.info("Proceeding with downstream tasks...")
- return condition
-
- if not self.downstream_task_ids:
- self.log.info("No downstream tasks; nothing to do.")
- return condition
-
- dag_run = context["dag_run"]
-
- def get_tasks_to_skip():
- if self.ignore_downstream_trigger_rules is True:
- tasks = context["task"].get_flat_relatives(upstream=False)
- else:
- tasks = context["task"].get_direct_relatives(upstream=False)
- for t in tasks:
- if not t.is_teardown:
- yield t
-
- to_skip = get_tasks_to_skip()
-
- # this let's us avoid an intermediate list unless debug logging
- if self.log.getEffectiveLevel() <= logging.DEBUG:
- self.log.debug("Downstream task IDs %s", to_skip :=
list(get_tasks_to_skip()))
-
- self.log.info("Skipping downstream tasks")
if AIRFLOW_V_3_0_PLUS:
- self.skip(
- dag_id=dag_run.dag_id,
- run_id=dag_run.run_id,
- tasks=to_skip,
- map_index=context["ti"].map_index,
- )
+ raise SkipDownstreamTaskInstances(condition=condition,
+
ignore_downstream_trigger_rules=self.ignore_downstream_trigger_rules
+ )
Review Comment:
I think that it makes more sense in the new abstraction to handle the
"skipping" logic in the task runner rather than the standard operator itself,
since the operator cannot set XComs directly.
The operator runs some logic which results in a boolean value, and upon
signaling the task runner - the latter would handle the setting the xcom with a
list of all affected tasks (it already has all of the required information for
that, except for the condition and the configruation).
--
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.
To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
