You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by po...@apache.org on 2024/07/26 10:04:39 UTC
(airflow) branch main updated: Revert "Refactor `BaseSensorOperator` introduce `skip_policy` parameter (#40924)" (#41045)
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new e0ccaa96e2 Revert "Refactor `BaseSensorOperator` introduce `skip_policy` parameter (#40924)" (#41045)
e0ccaa96e2 is described below
commit e0ccaa96e238821168aa05cdcd49d94884fa7948
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Fri Jul 26 12:04:32 2024 +0200
Revert "Refactor `BaseSensorOperator` introduce `skip_policy` parameter (#40924)" (#41045)
This reverts commit ba78d54538b2c27bc607bc760308ecc6c359c35d.
This is needed because provider tests started to fail and this one
should be combined and tested wiht #40923 including compatibility.
---
airflow/example_dags/example_sensors.py | 22 ++-----
airflow/sensors/base.py | 99 +++++++-----------------------
tests/decorators/test_sensor.py | 10 +--
tests/sensors/test_base.py | 45 ++++++--------
tests/sensors/test_external_task_sensor.py | 29 +++++----
5 files changed, 65 insertions(+), 140 deletions(-)
diff --git a/airflow/example_dags/example_sensors.py b/airflow/example_dags/example_sensors.py
index 3925a518d6..9dbe83d6e4 100644
--- a/airflow/example_dags/example_sensors.py
+++ b/airflow/example_dags/example_sensors.py
@@ -23,7 +23,6 @@ import pendulum
from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
-from airflow.sensors.base import SkipPolicy
from airflow.sensors.bash import BashSensor
from airflow.sensors.filesystem import FileSensor
from airflow.sensors.python import PythonSensor
@@ -69,7 +68,7 @@ with DAG(
t2 = TimeSensor(
task_id="timeout_after_second_date_in_the_future",
timeout=1,
- skip_policy=SkipPolicy.SKIP_ON_SOFT_ERROR,
+ soft_fail=True,
target_time=(datetime.datetime.now(tz=datetime.timezone.utc) + datetime.timedelta(hours=1)).time(),
)
# [END example_time_sensors]
@@ -82,7 +81,7 @@ with DAG(
t2a = TimeSensorAsync(
task_id="timeout_after_second_date_in_the_future_async",
timeout=1,
- skip_policy=SkipPolicy.SKIP_ON_SOFT_ERROR,
+ soft_fail=True,
target_time=(datetime.datetime.now(tz=datetime.timezone.utc) + datetime.timedelta(hours=1)).time(),
)
# [END example_time_sensors_async]
@@ -90,12 +89,7 @@ with DAG(
# [START example_bash_sensors]
t3 = BashSensor(task_id="Sensor_succeeds", bash_command="exit 0")
- t4 = BashSensor(
- task_id="Sensor_fails_after_3_seconds",
- timeout=3,
- skip_policy=SkipPolicy.SKIP_ON_SOFT_ERROR,
- bash_command="exit 1",
- )
+ t4 = BashSensor(task_id="Sensor_fails_after_3_seconds", timeout=3, soft_fail=True, bash_command="exit 1")
# [END example_bash_sensors]
t5 = BashOperator(task_id="remove_file", bash_command="rm -rf /tmp/temporary_file_for_testing")
@@ -118,19 +112,13 @@ with DAG(
t9 = PythonSensor(task_id="success_sensor_python", python_callable=success_callable)
t10 = PythonSensor(
- task_id="failure_timeout_sensor_python",
- timeout=3,
- skip_policy=SkipPolicy.SKIP_ON_SOFT_ERROR,
- python_callable=failure_callable,
+ task_id="failure_timeout_sensor_python", timeout=3, soft_fail=True, python_callable=failure_callable
)
# [END example_python_sensors]
# [START example_day_of_week_sensor]
t11 = DayOfWeekSensor(
- task_id="week_day_sensor_failing_on_timeout",
- timeout=3,
- skip_policy=SkipPolicy.SKIP_ON_SOFT_ERROR,
- week_day=WeekDay.MONDAY,
+ task_id="week_day_sensor_failing_on_timeout", timeout=3, soft_fail=True, week_day=WeekDay.MONDAY
)
# [END example_day_of_week_sensor]
diff --git a/airflow/sensors/base.py b/airflow/sensors/base.py
index a20aeaabc0..7df76fae52 100644
--- a/airflow/sensors/base.py
+++ b/airflow/sensors/base.py
@@ -18,12 +18,10 @@
from __future__ import annotations
import datetime
-import enum
import functools
import hashlib
import time
import traceback
-import warnings
from datetime import timedelta
from typing import TYPE_CHECKING, Any, Callable, Iterable
@@ -39,7 +37,6 @@ from airflow.exceptions import (
AirflowSensorTimeout,
AirflowSkipException,
AirflowTaskTimeout,
- RemovedInAirflow3Warning,
TaskDeferralError,
)
from airflow.executors.executor_loader import ExecutorLoader
@@ -54,7 +51,6 @@ from airflow.utils import timezone
# See https://github.com/apache/airflow/issues/16035
from airflow.utils.decorators import apply_defaults # noqa: F401
from airflow.utils.session import NEW_SESSION, provide_session
-from airflow.utils.types import NOTSET, ArgNotSet
if TYPE_CHECKING:
from sqlalchemy.orm.session import Session
@@ -118,24 +114,6 @@ def _orig_start_date(
)
-class SkipPolicy(str, enum.Enum):
- """Class with sensor's skip policies."""
-
- # if poke method raise an exception, sensor will not be skipped on.
- NONE = "none"
-
- # If poke method raises an exception, sensor will be skipped on.
- SKIP_ON_ANY_ERROR = "skip_on_any_error"
-
- # If poke method raises AirflowSensorTimeout, AirflowTaskTimeout, AirflowFailException
- # sensor will be skipped on.
- SKIP_ON_SOFT_ERROR = "skip_on_soft_error"
-
- # If poke method raises an exception different from AirflowSensorTimeout, AirflowTaskTimeout,
- # AirflowSkipException, sensor will ignore exception and re-poke until timeout.
- IGNORE_ERROR = "ignore_error"
-
-
class BaseSensorOperator(BaseOperator, SkipMixin):
"""
Sensor operators are derived from this class and inherit these attributes.
@@ -143,8 +121,8 @@ class BaseSensorOperator(BaseOperator, SkipMixin):
Sensor operators keep executing at a time interval and succeed when
a criteria is met and fail if and when they time out.
- :param soft_fail: deprecated parameter same effect than SkipPolicy.SKIP_ON_SOFT_ERROR
- Mutually exclusive with skip_policy and silent_fail.
+ :param soft_fail: Set to true to mark the task as SKIPPED on failure.
+ Mutually exclusive with never_fail.
:param poke_interval: Time that the job should wait in between each try.
Can be ``timedelta`` or ``float`` seconds.
:param timeout: Time elapsed before the task times out and fails.
@@ -172,13 +150,13 @@ class BaseSensorOperator(BaseOperator, SkipMixin):
:param exponential_backoff: allow progressive longer waits between
pokes by using exponential backoff algorithm
:param max_wait: maximum wait interval between pokes, can be ``timedelta`` or ``float`` seconds
- :param silent_fail: deprecated parameter same effect than SkipPolicy.IGNORE_ERROR
- Mutually exclusive with skip_policy and soft_fail.
- :param skip_policy: defines the rule by which sensor skip itself. Options are:
- ``{ none | skip_on_any_error | skip_on_soft_error | ignore_error }``
- default is ``none``. Options can be set as string or
- using the constants defined in the static class ``airflow.sensors.base.SkipPolicy``
- Mutually exclusive with soft_fail and silent_fail.
+ :param silent_fail: If true, and poke method raises an exception different from
+ AirflowSensorTimeout, AirflowTaskTimeout, AirflowSkipException
+ and AirflowFailException, the sensor will log the error and continue
+ its execution. Otherwise, the sensor task fails, and it can be retried
+ based on the provided `retries` parameter.
+ :param never_fail: If true, and poke method raises an exception, sensor will be skipped.
+ Mutually exclusive with soft_fail.
"""
ui_color: str = "#e6f1f2"
@@ -198,7 +176,7 @@ class BaseSensorOperator(BaseOperator, SkipMixin):
exponential_backoff: bool = False,
max_wait: timedelta | float | None = None,
silent_fail: bool = False,
- skip_policy: str | ArgNotSet = NOTSET, # SkipPolicy.NONE,
+ never_fail: bool = False,
**kwargs,
) -> None:
super().__init__(**kwargs)
@@ -208,44 +186,11 @@ class BaseSensorOperator(BaseOperator, SkipMixin):
self.mode = mode
self.exponential_backoff = exponential_backoff
self.max_wait = self._coerce_max_wait(max_wait)
- if skip_policy != NOTSET:
- if sum([soft_fail, silent_fail]) > 0:
- raise ValueError(
- "skip_policy and deprecated soft_fail and silent_fail parameters are mutually exclusive."
- )
-
- if skip_policy == SkipPolicy.SKIP_ON_SOFT_ERROR:
- self.soft_fail = True
- elif skip_policy == SkipPolicy.IGNORE_ERROR:
- self.silent_fail = True
- else:
- if sum([soft_fail, silent_fail]) > 1:
- raise ValueError(
- "soft_fail and silent_fail are mutually exclusive, you can not provide more than one."
- )
-
- if soft_fail:
- warnings.warn(
- "`soft_fail` is deprecated and will be removed in a future version. "
- "Please provide skip_policy=SkipPolicy.skip_on_soft_error instead.",
- RemovedInAirflow3Warning,
- stacklevel=3,
- )
- skip_policy = SkipPolicy.SKIP_ON_SOFT_ERROR
- elif silent_fail:
- warnings.warn(
- "`silent_fail` is deprecated and will be removed in a future version. "
- "Please provide skip_policy=SkipPolicy.IGNORE_ERRORS instead.",
- RemovedInAirflow3Warning,
- stacklevel=3,
- )
- skip_policy = SkipPolicy.IGNORE_ERROR
-
- else:
- skip_policy = SkipPolicy.NONE
+ if soft_fail is True and never_fail is True:
+ raise ValueError("soft_fail and never_fail are mutually exclusive, you can not provide both.")
self.silent_fail = silent_fail
- self.skip_policy = skip_policy
+ self.never_fail = never_fail
self._validate_input_values()
@staticmethod
@@ -344,19 +289,19 @@ class BaseSensorOperator(BaseOperator, SkipMixin):
AirflowTaskTimeout,
AirflowFailException,
) as e:
- if self.skip_policy == SkipPolicy.SKIP_ON_SOFT_ERROR:
- raise AirflowSkipException("Skipping due skip_policy set to skip_on_soft_error.") from e
- elif self.skip_policy == SkipPolicy.SKIP_ON_ANY_ERROR:
- raise AirflowSkipException("Skipping due skip_policy set to SKIP_ON_ANY_ERROR.") from e
+ if self.soft_fail:
+ raise AirflowSkipException("Skipping due to soft_fail is set to True.") from e
+ elif self.never_fail:
+ raise AirflowSkipException("Skipping due to never_fail is set to True.") from e
raise e
except AirflowSkipException as e:
raise e
except Exception as e:
- if self.skip_policy == SkipPolicy.IGNORE_ERROR:
+ if self.silent_fail:
self.log.error("Sensor poke failed: \n %s", traceback.format_exc())
poke_return = False
- elif self.skip_policy == SkipPolicy.SKIP_ON_ANY_ERROR:
- raise AirflowSkipException("Skipping due to SKIP_ON_ANY_ERROR is set to True.") from e
+ elif self.never_fail:
+ raise AirflowSkipException("Skipping due to never_fail is set to True.") from e
else:
raise e
@@ -372,7 +317,7 @@ class BaseSensorOperator(BaseOperator, SkipMixin):
f"the specified timeout of {self.timeout}."
)
- if self.skip_policy == SkipPolicy.SKIP_ON_SOFT_ERROR:
+ if self.soft_fail:
raise AirflowSkipException(message)
else:
raise AirflowSensorTimeout(message)
@@ -395,7 +340,7 @@ class BaseSensorOperator(BaseOperator, SkipMixin):
try:
return super().resume_execution(next_method, next_kwargs, context)
except (AirflowException, TaskDeferralError) as e:
- if self.skip_policy == SkipPolicy.SKIP_ON_SOFT_ERROR:
+ if self.soft_fail:
raise AirflowSkipException(str(e)) from e
raise
diff --git a/tests/decorators/test_sensor.py b/tests/decorators/test_sensor.py
index 7dee0f6865..77852f34f7 100644
--- a/tests/decorators/test_sensor.py
+++ b/tests/decorators/test_sensor.py
@@ -23,7 +23,7 @@ import pytest
from airflow.decorators import task
from airflow.exceptions import AirflowSensorTimeout
from airflow.models import XCom
-from airflow.sensors.base import PokeReturnValue, SkipPolicy
+from airflow.sensors.base import PokeReturnValue
from airflow.utils.state import State
pytestmark = pytest.mark.db_test
@@ -141,8 +141,8 @@ class TestSensorDecorator:
if ti.task_id == "dummy_f":
assert ti.state == State.NONE
- def test_basic_sensor_skip_on_soft_error(self, dag_maker):
- @task.sensor(timeout=0, skip_policy=SkipPolicy.SKIP_ON_SOFT_ERROR)
+ def test_basic_sensor_soft_fail(self, dag_maker):
+ @task.sensor(timeout=0, soft_fail=True)
def sensor_f():
return PokeReturnValue(is_done=False, xcom_value="xcom_value")
@@ -165,8 +165,8 @@ class TestSensorDecorator:
if ti.task_id == "dummy_f":
assert ti.state == State.NONE
- def test_basic_sensor_skip_on_soft_error_returns_bool(self, dag_maker):
- @task.sensor(timeout=0, skip_policy=SkipPolicy.SKIP_ON_SOFT_ERROR)
+ def test_basic_sensor_soft_fail_returns_bool(self, dag_maker):
+ @task.sensor(timeout=0, soft_fail=True)
def sensor_f():
return False
diff --git a/tests/sensors/test_base.py b/tests/sensors/test_base.py
index 7660effc7c..79b88eb40d 100644
--- a/tests/sensors/test_base.py
+++ b/tests/sensors/test_base.py
@@ -51,7 +51,7 @@ from airflow.providers.celery.executors.celery_executor import CeleryExecutor
from airflow.providers.celery.executors.celery_kubernetes_executor import CeleryKubernetesExecutor
from airflow.providers.cncf.kubernetes.executors.kubernetes_executor import KubernetesExecutor
from airflow.providers.cncf.kubernetes.executors.local_kubernetes_executor import LocalKubernetesExecutor
-from airflow.sensors.base import BaseSensorOperator, PokeReturnValue, SkipPolicy, poke_mode_only
+from airflow.sensors.base import BaseSensorOperator, PokeReturnValue, poke_mode_only
from airflow.ti_deps.deps.ready_to_reschedule import ReadyToRescheduleDep
from airflow.utils import timezone
from airflow.utils.session import create_session
@@ -178,8 +178,8 @@ class TestBaseSensor:
if ti.task_id == DUMMY_OP:
assert ti.state == State.NONE
- def test_skip_on_soft_error(self, make_sensor):
- sensor, dr = make_sensor(False, skip_policy=SkipPolicy.SKIP_ON_SOFT_ERROR)
+ def test_soft_fail(self, make_sensor):
+ sensor, dr = make_sensor(False, soft_fail=True)
self._run(sensor)
tis = dr.get_task_instances()
@@ -194,8 +194,8 @@ class TestBaseSensor:
"exception_cls",
(ValueError,),
)
- def test_skip_on_soft_error_with_exception(self, make_sensor, exception_cls):
- sensor, dr = make_sensor(False, skip_policy=SkipPolicy.SKIP_ON_SOFT_ERROR)
+ def test_soft_fail_with_exception(self, make_sensor, exception_cls):
+ sensor, dr = make_sensor(False, soft_fail=True)
sensor.poke = Mock(side_effect=[exception_cls(None)])
with pytest.raises(ValueError):
self._run(sensor)
@@ -216,8 +216,8 @@ class TestBaseSensor:
AirflowFailException,
),
)
- def test_skip_on_soft_error_with_skip_exception(self, make_sensor, exception_cls):
- sensor, dr = make_sensor(False, skip_policy=SkipPolicy.SKIP_ON_SOFT_ERROR)
+ def test_soft_fail_with_skip_exception(self, make_sensor, exception_cls):
+ sensor, dr = make_sensor(False, soft_fail=True)
sensor.poke = Mock(side_effect=[exception_cls(None)])
self._run(sensor)
@@ -233,8 +233,8 @@ class TestBaseSensor:
"exception_cls",
(AirflowSensorTimeout, AirflowTaskTimeout, AirflowFailException, Exception),
)
- def test_skip_on_any_error_with_skip_exception(self, make_sensor, exception_cls):
- sensor, dr = make_sensor(False, skip_policy=SkipPolicy.SKIP_ON_ANY_ERROR)
+ def test_never_fail_with_skip_exception(self, make_sensor, exception_cls):
+ sensor, dr = make_sensor(False, never_fail=True)
sensor.poke = Mock(side_effect=[exception_cls(None)])
self._run(sensor)
@@ -246,12 +246,9 @@ class TestBaseSensor:
if ti.task_id == DUMMY_OP:
assert ti.state == State.NONE
- def test_skip_on_soft_error_with_retries(self, make_sensor):
+ def test_soft_fail_with_retries(self, make_sensor):
sensor, dr = make_sensor(
- return_value=False,
- skip_policy=SkipPolicy.SKIP_ON_SOFT_ERROR,
- retries=1,
- retry_delay=timedelta(milliseconds=1),
+ return_value=False, soft_fail=True, retries=1, retry_delay=timedelta(milliseconds=1)
)
# first run times out and task instance is skipped
@@ -359,13 +356,9 @@ class TestBaseSensor:
assert sensor_ti.state == State.FAILED
assert dummy_ti.state == State.NONE
- def test_skip_on_soft_error_with_reschedule(self, make_sensor, time_machine, session):
+ def test_soft_fail_with_reschedule(self, make_sensor, time_machine, session):
sensor, dr = make_sensor(
- return_value=False,
- poke_interval=10,
- timeout=5,
- skip_policy=SkipPolicy.SKIP_ON_SOFT_ERROR,
- mode="reschedule",
+ return_value=False, poke_interval=10, timeout=5, soft_fail=True, mode="reschedule"
)
def _get_tis():
@@ -917,7 +910,7 @@ class TestBaseSensor:
retries=2,
retry_delay=timedelta(seconds=3),
mode="reschedule",
- skip_policy=SkipPolicy.IGNORE_ERROR,
+ silent_fail=True,
)
def _get_sensor_ti():
@@ -1117,14 +1110,14 @@ class TestPokeModeOnly:
class TestAsyncSensor:
@pytest.mark.parametrize(
- "skip_policy, expected_exception",
+ "soft_fail, expected_exception",
[
- (SkipPolicy.SKIP_ON_SOFT_ERROR, AirflowSkipException),
- (SkipPolicy.NONE, AirflowException),
+ (True, AirflowSkipException),
+ (False, AirflowException),
],
)
- def test_fail_after_resuming_deferred_sensor(self, skip_policy, expected_exception):
- async_sensor = DummyAsyncSensor(task_id="dummy_async_sensor", skip_policy=skip_policy)
+ def test_fail_after_resuming_deferred_sensor(self, soft_fail, expected_exception):
+ async_sensor = DummyAsyncSensor(task_id="dummy_async_sensor", soft_fail=soft_fail)
ti = TaskInstance(task=async_sensor)
ti.next_method = "execute_complete"
with pytest.raises(expected_exception):
diff --git a/tests/sensors/test_external_task_sensor.py b/tests/sensors/test_external_task_sensor.py
index d933a4c226..58e25a3de0 100644
--- a/tests/sensors/test_external_task_sensor.py
+++ b/tests/sensors/test_external_task_sensor.py
@@ -38,7 +38,6 @@ from airflow.models.xcom_arg import XComArg
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
from airflow.operators.python import PythonOperator
-from airflow.sensors.base import SkipPolicy
from airflow.sensors.external_task import (
ExternalTaskMarker,
ExternalTaskSensor,
@@ -330,7 +329,7 @@ class TestExternalTaskSensor:
f"Poking for tasks ['{TEST_TASK_ID}'] in dag {TEST_DAG_ID} on {DEFAULT_DATE.isoformat()} ... "
) in caplog.messages
- def test_external_task_sensor_skip_on_soft_error_failed_states_as_skipped(self):
+ def test_external_task_sensor_soft_fail_failed_states_as_skipped(self):
self.add_time_sensor()
op = ExternalTaskSensor(
task_id="test_external_task_sensor_check",
@@ -338,7 +337,7 @@ class TestExternalTaskSensor:
external_task_id=TEST_TASK_ID,
allowed_states=[State.FAILED],
failed_states=[State.SUCCESS],
- skip_policy=SkipPolicy.SKIP_ON_SOFT_ERROR,
+ soft_fail=True,
dag=self.dag,
)
@@ -468,7 +467,7 @@ class TestExternalTaskSensor:
op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, ignore_ti_state=True)
assert (f"Poking for DAG 'other_dag' on {DEFAULT_DATE.isoformat()} ... ") in caplog.messages
- def test_external_dag_sensor_skip_on_soft_error_as_skipped(self):
+ def test_external_dag_sensor_soft_fail_as_skipped(self):
other_dag = DAG("other_dag", default_args=self.args, end_date=DEFAULT_DATE, schedule="@once")
other_dag.create_dagrun(
run_id="test",
@@ -483,7 +482,7 @@ class TestExternalTaskSensor:
external_task_id=None,
allowed_states=[State.FAILED],
failed_states=[State.SUCCESS],
- skip_policy=SkipPolicy.SKIP_ON_SOFT_ERROR,
+ soft_fail=True,
dag=self.dag,
)
@@ -881,14 +880,14 @@ exit 0
),
)
@pytest.mark.parametrize(
- "skip_policy, expected_exception",
+ "soft_fail, expected_exception",
(
(
- SkipPolicy.NONE,
+ False,
AirflowException,
),
(
- SkipPolicy.SKIP_ON_SOFT_ERROR,
+ True,
AirflowSkipException,
),
),
@@ -896,7 +895,7 @@ exit 0
@mock.patch("airflow.sensors.external_task.ExternalTaskSensor.get_count")
@mock.patch("airflow.sensors.external_task.ExternalTaskSensor._get_dttm_filter")
def test_fail_poke(
- self, _get_dttm_filter, get_count, skip_policy, expected_exception, kwargs, expected_message
+ self, _get_dttm_filter, get_count, soft_fail, expected_exception, kwargs, expected_message
):
_get_dttm_filter.return_value = []
get_count.return_value = 1
@@ -905,7 +904,7 @@ exit 0
external_dag_id=TEST_DAG_ID,
allowed_states=["success"],
dag=self.dag,
- skip_policy=skip_policy,
+ soft_fail=soft_fail,
deferrable=False,
**kwargs,
)
@@ -938,14 +937,14 @@ exit 0
),
)
@pytest.mark.parametrize(
- "skip_policy, expected_exception",
+ "soft_fail, expected_exception",
(
(
- SkipPolicy.NONE,
+ False,
AirflowException,
),
(
- SkipPolicy.SKIP_ON_SOFT_ERROR,
+ True,
AirflowException,
),
),
@@ -960,7 +959,7 @@ exit 0
exists,
get_dag,
_get_dttm_filter,
- skip_policy,
+ soft_fail,
expected_exception,
response_get_current,
response_exists,
@@ -979,7 +978,7 @@ exit 0
external_dag_id=TEST_DAG_ID,
allowed_states=["success"],
dag=self.dag,
- skip_policy=skip_policy,
+ soft_fail=soft_fail,
check_existence=True,
**kwargs,
)