[GitHub] [airflow] randr97 commented on a change in pull request #8509: [AIRFLOW-4568]The ExternalTaskSensor should be configurable to raise an Airflow Exception in case the poked external task reaches a disallowed state, such as f.i. failed

[GitBox]

randr97 commented on a change in pull request #8509:
URL: https://github.com/apache/airflow/pull/8509#discussion_r414575632



##########
File path: airflow/sensors/external_task_sensor.py
##########
@@ -68,23 +70,35 @@ def __init__(self,
                  external_dag_id,
                  external_task_id=None,
                  allowed_states=None,
+                 failed_states=None,
                  execution_delta=None,
                  execution_date_fn=None,
                  check_existence=False,
                  *args,
                  **kwargs):
         super().__init__(*args, **kwargs)
         self.allowed_states = allowed_states or [State.SUCCESS]
+        self.failed_states = failed_states or []
+
+        total_states = []
+        total_states.extend(self.allowed_states)
+        total_states.extend(self.failed_states)
+
+        if len(list(set(total_states))) < (len(self.failed_states) + 
len(self.allowed_states)):
+            raise AirflowException("Duplicate values provided as allowed "
+                                   "`{}` and failed states `{}`"
+                                   .format(self.allowed_states, 
self.failed_states))
+
         if external_task_id:
-            if not set(self.allowed_states) <= set(State.task_states):
+            if not set(total_states) <= set(State.task_states):
                 raise ValueError(
-                    'Valid values for `allowed_states` '
+                    'Valid values for `allowed_states` and `failed_states` '
                     'when `external_task_id` is not `None`: 
{}'.format(State.task_states)
                 )
         else:
-            if not set(self.allowed_states) <= set(State.dag_states):
+            if not set(total_states) <= set(State.dag_states):
                 raise ValueError(
-                    'Valid values for `allowed_states` '
+                    'Valid values for `allowed_states` and `failed_states` '
                     'when `external_task_id` is `None`: 
{}'.format(State.dag_states)
                 )

Review comment:
       `error_template = 'Valid values for `allowed_states` and `failed_states` 
'\
                    'when `external_task_id` is not `None`: {}'
   if external_task_id and not set(total_states) <= set(State.task_states):
       raise ValueError(error_template.format(State.task_states))
   elif not set(total_states) <= set(State.dag_states):
       raise ValueError(error_template.format(State.dag_states))`
   Jus my personal view :P




----------------------------------------------------------------
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:
us...@infra.apache.org