diff --git a/providers/tableau/docs/operators.rst b/providers/tableau/docs/operators.rst index 9bcc830b76184..ab6141ce3dd64 100644 --- a/providers/tableau/docs/operators.rst +++ b/providers/tableau/docs/operators.rst @@ -39,6 +39,7 @@ Using the Operator | **timeout**: maximum total time in seconds to wait for a blocking refresh before giving up with a ``TimeoutError``. **float** - Default: **None** (wait indefinitely) | **exponential_backoff**: grow the wait between status checks by 50% each time, starting from ``check_interval``, instead of staying fixed. **bool** - Default: **False** | **max_check_interval**: maximum interval in seconds between two consecutive status checks when ``exponential_backoff`` is enabled. **float** - Default: **None** (uncapped) +| **skip_on_conflict**: When ``True``, treat a Tableau ``409093 Resource Conflict`` error while triggering a refresh or running an extract refresh task (a refresh/run for the same resource is already queued or running) as a skipped task instead of a failure. Applies when ``method="refresh"`` and when ``method="run"`` on ``tasks``. **bool** - Default: **False** | **tableau_conn_id**: The credentials to authenticate to the Tableau Server. **str** - Default: **tableau_default** | | diff --git a/providers/tableau/src/airflow/providers/tableau/operators/tableau.py b/providers/tableau/src/airflow/providers/tableau/operators/tableau.py index f7158f95e4226..bcf09798f7c6d 100644 --- a/providers/tableau/src/airflow/providers/tableau/operators/tableau.py +++ b/providers/tableau/src/airflow/providers/tableau/operators/tableau.py @@ -19,11 +19,12 @@ from collections.abc import Sequence from typing import TYPE_CHECKING -from tableauserverclient import JobItem +from tableauserverclient import JobItem, ServerResponseError from airflow.providers.common.compat.sdk import ( AirflowException, AirflowOptionalProviderFeatureException, + AirflowSkipException, BaseOperator, ) from airflow.providers.tableau.hooks.tableau import ( @@ -35,6 +36,12 @@ if TYPE_CHECKING: from airflow.providers.common.compat.sdk import Context +# Tableau REST API error code returned when an extract refresh -- triggered directly via +# datasources/workbooks "refresh" or via running a scheduled extract refresh task ("tasks.run") +# -- is requested for a resource that already has one queued or running. +# See: https://help.tableau.com/current/api/rest_api/en-us/REST/rest_api_concepts_errors.htm +RESOURCE_CONFLICT_ERROR_CODE = "409093" + RESOURCES_METHODS = { "datasources": ["delete", "refresh"], "groups": ["delete"], @@ -75,6 +82,11 @@ class TableauOperator(BaseOperator): when ``exponential_backoff`` is enabled. ``None`` leaves the growth uncapped. :param incremental_refresh: Whether to perform an incremental refresh instead of a full refresh. Only applies to datasource and workbook refresh operations. Defaults to False (full refresh). + :param skip_on_conflict: When ``True``, treat a Tableau ``409093 Resource Conflict`` error + while triggering a refresh, or while running an extract refresh task, as a skipped task + instead of a failure. Raised when a refresh/run for the same resource is already queued + or running. Applies to ``method="refresh"`` and to ``method="run"`` on ``tasks``. + Defaults to ``False``, so the conflict fails the task as before. :param tableau_conn_id: The :ref:`Tableau Connection id ` containing the credentials to authenticate to the Tableau Server. """ @@ -98,6 +110,7 @@ def __init__( exponential_backoff: bool = False, max_check_interval: float | None = None, incremental_refresh: bool = False, + skip_on_conflict: bool = False, tableau_conn_id: str = "tableau_default", **kwargs, ) -> None: @@ -113,6 +126,7 @@ def __init__( self.site_id = site_id self.blocking_refresh = blocking_refresh self.incremental_refresh = incremental_refresh + self.skip_on_conflict = skip_on_conflict self.tableau_conn_id = tableau_conn_id def execute(self, context: Context) -> str: @@ -147,24 +161,40 @@ def execute(self, context: Context) -> str: if self.resource == "tasks" and self.method == "run": task_item = resource.get_by_id(resource_id) - response_bytes = method(task_item) + try: + response_bytes = method(task_item) + except ServerResponseError as e: + if self.skip_on_conflict and e.code == RESOURCE_CONFLICT_ERROR_CODE: + raise AirflowSkipException( + f"Tableau task {resource_id} run is already queued or running " + f"({e.code}: {e.summary}); skipping this task." + ) from e + raise job_items = JobItem.from_response(response_bytes, tableau_hook.server.namespace) if not job_items: raise ValueError("Tableau tasks.run returned no JobItem in response") job_id = job_items[0].id elif self.method == "refresh": - if self.incremental_refresh: - try: - response = method(resource_id, incremental=True) - except TypeError as e: - if "incremental" in str(e): - raise AirflowOptionalProviderFeatureException( - "Incremental refresh requires tableauserverclient>=0.35. " - "Please upgrade: pip install 'tableauserverclient>=0.35'" - ) from e - raise - else: - response = method(resource_id) + try: + if self.incremental_refresh: + try: + response = method(resource_id, incremental=True) + except TypeError as e: + if "incremental" in str(e): + raise AirflowOptionalProviderFeatureException( + "Incremental refresh requires tableauserverclient>=0.35. " + "Please upgrade: pip install 'tableauserverclient>=0.35'" + ) from e + raise + else: + response = method(resource_id) + except ServerResponseError as e: + if self.skip_on_conflict and e.code == RESOURCE_CONFLICT_ERROR_CODE: + raise AirflowSkipException( + f"Tableau {self.resource} refresh is already queued or running " + f"({e.code}: {e.summary}); skipping this task." + ) from e + raise job_id = response.id else: response = method(resource_id) diff --git a/providers/tableau/tests/unit/tableau/operators/test_tableau.py b/providers/tableau/tests/unit/tableau/operators/test_tableau.py index 759a0344c0168..6b27de71d1eb4 100644 --- a/providers/tableau/tests/unit/tableau/operators/test_tableau.py +++ b/providers/tableau/tests/unit/tableau/operators/test_tableau.py @@ -19,8 +19,13 @@ from unittest.mock import Mock, patch import pytest +from tableauserverclient import ServerResponseError -from airflow.providers.common.compat.sdk import AirflowException, AirflowOptionalProviderFeatureException +from airflow.providers.common.compat.sdk import ( + AirflowException, + AirflowOptionalProviderFeatureException, + AirflowSkipException, +) from airflow.providers.tableau.hooks.tableau import TableauJobFinishCode from airflow.providers.tableau.operators.tableau import TableauOperator @@ -530,3 +535,144 @@ def test_blocking_refresh_forwards_wait_for_state_options(self, mock_tableau_hoo exponential_backoff=True, max_check_interval=120, ) + + @patch("airflow.providers.tableau.operators.tableau.TableauHook") + def test_resource_refresh_skip_on_conflict_raises_skip_exception(self, mock_tableau_hook): + """ + Test that a 409093 Resource Conflict on a Tableau resource refresh is turned into an + AirflowSkipException when skip_on_conflict=True. + """ + mock_tableau_hook.get_all = Mock(return_value=self.mock_datasources) + mock_tableau_hook.return_value.__enter__ = Mock(return_value=mock_tableau_hook) + mock_tableau_hook.server.datasources.refresh.side_effect = ServerResponseError( + "409093", "Resource Conflict", "Job is already queued. Not queuing a duplicate." + ) + + operator = TableauOperator( + find="ds_2", + resource="datasources", + skip_on_conflict=True, + **self.kwargs, + ) + + with pytest.raises(AirflowSkipException): + operator.execute(context={}) + + @patch("airflow.providers.tableau.operators.tableau.TableauHook") + def test_incremental_resource_refresh_skip_on_conflict_raises_skip_exception(self, mock_tableau_hook): + """ + Test that a 409093 Resource Conflict on a Tableau incremental resource refresh is turned into an + AirflowSkipException when skip_on_conflict=True. + """ + mock_tableau_hook.get_all = Mock(return_value=self.mock_datasources) + mock_tableau_hook.return_value.__enter__ = Mock(return_value=mock_tableau_hook) + mock_tableau_hook.server.datasources.refresh.side_effect = ServerResponseError( + "409093", "Resource Conflict", "Job is already queued. Not queuing a duplicate." + ) + operator = TableauOperator( + find="ds_2", + resource="datasources", + incremental_refresh=True, + skip_on_conflict=True, + **self.kwargs, + ) + with pytest.raises(AirflowSkipException): + operator.execute(context={}) + + @patch("airflow.providers.tableau.operators.tableau.TableauHook") + def test_resource_refresh_conflict_not_skipped_by_default(self, mock_tableau_hook): + """ + Test that a 409093 Resource Conflict on a Tableau resource refresh still fails the + task when skip_on_conflict is left at its default (False), preserving pre-existing + behavior. + """ + mock_tableau_hook.get_all = Mock(return_value=self.mock_datasources) + mock_tableau_hook.return_value.__enter__ = Mock(return_value=mock_tableau_hook) + mock_tableau_hook.server.datasources.refresh.side_effect = ServerResponseError( + "409093", "Resource Conflict", "Job is already queued. Not queuing a duplicate." + ) + + operator = TableauOperator( + find="ds_2", + resource="datasources", + **self.kwargs, + ) + + with pytest.raises(ServerResponseError): + operator.execute(context={}) + + @patch("airflow.providers.tableau.operators.tableau.TableauHook") + def test_resource_refresh_skip_on_conflict_does_not_swallow_other_server_errors(self, mock_tableau_hook): + """ + Test that skip_on_conflict on a Tableau resource refresh only catches the 409093 + conflict code; other ServerResponseErrors still fail the task. + """ + mock_tableau_hook.get_all = Mock(return_value=self.mock_datasources) + mock_tableau_hook.return_value.__enter__ = Mock(return_value=mock_tableau_hook) + mock_tableau_hook.server.datasources.refresh.side_effect = ServerResponseError( + "500000", "Internal Server Error", "Something else went wrong." + ) + + operator = TableauOperator( + find="ds_2", + resource="datasources", + skip_on_conflict=True, + **self.kwargs, + ) + + with pytest.raises(ServerResponseError): + operator.execute(context={}) + + @patch("airflow.providers.tableau.operators.tableau.TableauHook") + def test_task_run_skip_on_conflict_raises_skip_exception(self, mock_tableau_hook): + """ + Test that a 409093 Resource Conflict on a Tableau task run is turned into an + AirflowSkipException when skip_on_conflict=True. + """ + mock_tableau_hook.return_value.__enter__ = Mock(return_value=mock_tableau_hook) + mock_tableau_hook.server.tasks.get_by_id = Mock(return_value=Mock()) + mock_tableau_hook.server.tasks.run.side_effect = ServerResponseError( + "409093", "Resource Conflict", "Job is already queued. Not queuing a duplicate." + ) + + kwargs = {**self.kwargs, "method": "run", "match_with": "id"} + operator = TableauOperator(find="task-abc", resource="tasks", skip_on_conflict=True, **kwargs) + + with pytest.raises(AirflowSkipException): + operator.execute(context={}) + + @patch("airflow.providers.tableau.operators.tableau.TableauHook") + def test_task_run_conflict_not_skipped_by_default(self, mock_tableau_hook): + """ + Test that a 409093 Resource Conflict on a Tableau task run still fails the task when + skip_on_conflict is left at its default (False), preserving pre-existing behavior. + """ + mock_tableau_hook.return_value.__enter__ = Mock(return_value=mock_tableau_hook) + mock_tableau_hook.server.tasks.get_by_id = Mock(return_value=Mock()) + mock_tableau_hook.server.tasks.run.side_effect = ServerResponseError( + "409093", "Resource Conflict", "Job is already queued. Not queuing a duplicate." + ) + + kwargs = {**self.kwargs, "method": "run", "match_with": "id"} + operator = TableauOperator(find="task-abc", resource="tasks", **kwargs) + + with pytest.raises(ServerResponseError): + operator.execute(context={}) + + @patch("airflow.providers.tableau.operators.tableau.TableauHook") + def test_task_run_skip_on_conflict_does_not_swallow_other_server_errors(self, mock_tableau_hook): + """ + Test that skip_on_conflict on a Tableau task run only catches the 409093 conflict + code; other ServerResponseErrors still fail the task. + """ + mock_tableau_hook.return_value.__enter__ = Mock(return_value=mock_tableau_hook) + mock_tableau_hook.server.tasks.get_by_id = Mock(return_value=Mock()) + mock_tableau_hook.server.tasks.run.side_effect = ServerResponseError( + "500000", "Internal Server Error", "Something else went wrong." + ) + + kwargs = {**self.kwargs, "method": "run", "match_with": "id"} + operator = TableauOperator(find="task-abc", resource="tasks", skip_on_conflict=True, **kwargs) + + with pytest.raises(ServerResponseError): + operator.execute(context={})