Skip Tableau refresh/task run on 409093 resource conflict

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**
 |
 |

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 <howto/connection:tableau>`
         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)

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={})