diff --git a/airflow/example_dags/example_sensors.py b/airflow/example_dags/example_sensors.py index 3925a518d64f7..9dbe83d6e4c40 100644 --- a/airflow/example_dags/example_sensors.py +++ b/airflow/example_dags/example_sensors.py @@ -23,7 +23,6 @@ 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 @@ def failure_callable(): 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 @@ def failure_callable(): 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 @@ def failure_callable(): # [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 @@ def failure_callable(): 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 a20aeaabc083a..7df76fae52883 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 @@ AirflowSensorTimeout, AirflowSkipException, AirflowTaskTimeout, - RemovedInAirflow3Warning, TaskDeferralError, ) from airflow.executors.executor_loader import ExecutorLoader @@ -54,7 +51,6 @@ # 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 @@ def __init__( 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 @@ def __init__( 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 @@ def run_duration() -> float: 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 @@ def run_duration() -> float: 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 @@ def resume_execution(self, next_method: str, next_kwargs: dict[str, Any] | None, 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 7dee0f6865122..77852f34f7262 100644 --- a/tests/decorators/test_sensor.py +++ b/tests/decorators/test_sensor.py @@ -23,7 +23,7 @@ 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 @@ def dummy_f(): 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 @@ def dummy_f(): 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 7660effc7cb68..79b88eb40dbe6 100644 --- a/tests/sensors/test_base.py +++ b/tests/sensors/test_base.py @@ -51,7 +51,7 @@ 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 @@ def test_fail(self, make_sensor): 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 @@ def test_skip_on_soft_error(self, make_sensor): "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 @@ def test_skip_on_soft_error_with_exception(self, make_sensor, exception_cls): 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 @@ def test_skip_on_soft_error_with_skip_exception(self, make_sensor, exception_cls "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 @@ def test_skip_on_any_error_with_skip_exception(self, make_sensor, exception_cls) 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 @@ def _get_tis(): 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 @@ def test_reschedule_and_retry_timeout_and_silent_fail(self, make_sensor, time_ma 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 @@ def test_poke_mode_only_bad_poke(self): 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 d933a4c2266ef..58e25a3de0d04 100644 --- a/tests/sensors/test_external_task_sensor.py +++ b/tests/sensors/test_external_task_sensor.py @@ -38,7 +38,6 @@ 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 @@ def test_external_task_sensor_failed_states_as_success(self, caplog): 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 @@ def test_external_task_sensor_skip_on_soft_error_failed_states_as_skipped(self): 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 @@ def test_external_dag_sensor_log(self, caplog): 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 @@ def test_external_dag_sensor_skip_on_soft_error_as_skipped(self): 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 @@ def test_external_task_group_when_there_is_no_TIs(self): ), ) @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 @@ def test_external_task_group_when_there_is_no_TIs(self): @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 @@ def test_fail_poke( 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 @@ def test_fail_poke( ), ) @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 @@ def test_fail__check_for_existence( exists, get_dag, _get_dttm_filter, - skip_policy, + soft_fail, expected_exception, response_get_current, response_exists, @@ -979,7 +978,7 @@ def test_fail__check_for_existence( external_dag_id=TEST_DAG_ID, allowed_states=["success"], dag=self.dag, - skip_policy=skip_policy, + soft_fail=soft_fail, check_existence=True, **kwargs, )