uranusjr commented on a change in pull request #17421:
URL: https://github.com/apache/airflow/pull/17421#discussion_r690936002
##########
File path: airflow/operators/python.py
##########
@@ -26,6 +26,11 @@
from textwrap import dedent
from typing import Any, Callable, Dict, Iterable, List, Optional, Union
+try:
+ from typing import Literal
+except ImportError:
+ from typing_extensions import Literal
Review comment:
There’s `airflow.utils.typing_compat` for this.
##########
File path: airflow/operators/python.py
##########
@@ -185,16 +190,28 @@ def execute(self, context: Dict):
class ShortCircuitOperator(PythonOperator, SkipMixin):
"""
Allows a workflow to continue only if a condition is met. Otherwise, the
- workflow "short-circuits" and downstream tasks are skipped.
+ workflow "short-circuits" and downstream tasks are skipped. The
short-circuiting can be configured to
+ operated in one of two modes: "hard" or "soft". In "hard" mode, the
default mode, all downstream tasks
+ are skipped without considering the ``trigger_rule`` defined for tasks.
In "soft" mode, the
+ ``trigger_rule`` is respected.
The ShortCircuitOperator is derived from the PythonOperator. It evaluates a
- condition and short-circuits the workflow if the condition is False. Any
- downstream tasks are marked with a state of "skipped". If the condition is
- True, downstream tasks proceed as normal.
+ condition and short-circuits the workflow if the condition is False. Any
downstream tasks are marked with
+ a state of "skipped" based on the short-circuiting mode configured. If the
condition is True, downstream
+ tasks proceed as normal.
+
+ The condition is determined by the result of ``python_callable``.
- The condition is determined by the result of `python_callable`.
+ :param mode: If set to "hard", all downstream tasks from this operator
task will be skipped. This is
+ the default behavior. If set to `False`, downstream tasks will be
skipped but the ``trigger_rule``
Review comment:
```suggestion
the default behavior. If set to "soft", downstream tasks will be
skipped but the ``trigger_rule``
```
##########
File path: airflow/operators/python.py
##########
@@ -203,13 +220,26 @@ def execute(self, context: Dict):
self.log.info('Proceeding with downstream tasks...')
return
- self.log.info('Skipping downstream tasks...')
-
downstream_tasks = context['task'].get_flat_relatives(upstream=False)
- self.log.debug("Downstream task_ids %s", downstream_tasks)
+ self.log.debug("Downstream tasks %s", downstream_tasks)
if downstream_tasks:
- self.skip(context['dag_run'], context['ti'].execution_date,
downstream_tasks)
+ dag_run = context["dag_run"]
+ execution_date = context["ti"].execution_date
+
+ if self.mode == "hard":
+ self.log.info("Skipping downstream tasks with using a hard
short...")
+ self.skip(dag_run, execution_date, downstream_tasks)
+ elif self.mode == "soft":
+ self.log.info("Skipping downstream tasks with using a soft
short...")
+ # Explicitly setting the state of the direct, downstream
task(s) to "skipped" and letting the
+ # Scheduler handle the remaining downstream task(s)
appropriately.
+ self.skip(dag_run, execution_date,
context["task"].get_direct_relatives(upstream=False))
+ else:
+ raise ValueError(
+ f"The provided short-circuiting mode value, '{self.mode}',
is not supported. "
+ + "Please use either 'hard' or 'soft'."
Review comment:
```suggestion
f"The provided short-circuiting mode value,
'{self.mode}', is not supported. "
f"Please use either 'hard' or 'soft'."
```
This is called [string literal
concatenation](https://docs.python.org/3/reference/lexical_analysis.html#string-literal-concatenation).
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
