diff --git a/airflow/providers/salesforce/sensors/tableau_job_status.py b/airflow/providers/salesforce/sensors/tableau_job_status.py index b1fb4a71bc5dc..09e2a373eaccb 100644 --- a/airflow/providers/salesforce/sensors/tableau_job_status.py +++ b/airflow/providers/salesforce/sensors/tableau_job_status.py @@ -17,7 +17,7 @@ import warnings -from airflow.providers.tableau.sensors.tableau_job_status import ( # noqa +from airflow.providers.tableau.sensors.tableau import ( # noqa TableauJobFailedException, TableauJobStatusSensor, ) diff --git a/airflow/providers/tableau/example_dags/example_tableau.py b/airflow/providers/tableau/example_dags/example_tableau.py index 0ad956a735f65..53aba4c074839 100644 --- a/airflow/providers/tableau/example_dags/example_tableau.py +++ b/airflow/providers/tableau/example_dags/example_tableau.py @@ -24,7 +24,7 @@ from airflow import DAG from airflow.providers.tableau.operators.tableau import TableauOperator -from airflow.providers.tableau.sensors.tableau_job_status import TableauJobStatusSensor +from airflow.providers.tableau.sensors.tableau import TableauJobStatusSensor with DAG( dag_id='example_tableau', diff --git a/airflow/providers/tableau/example_dags/example_tableau_refresh_workbook.py b/airflow/providers/tableau/example_dags/example_tableau_refresh_workbook.py index f607e7eaf366a..31579003b80aa 100644 --- a/airflow/providers/tableau/example_dags/example_tableau_refresh_workbook.py +++ b/airflow/providers/tableau/example_dags/example_tableau_refresh_workbook.py @@ -24,7 +24,7 @@ from airflow import DAG from airflow.providers.tableau.operators.tableau import TableauOperator -from airflow.providers.tableau.sensors.tableau_job_status import TableauJobStatusSensor +from airflow.providers.tableau.sensors.tableau import TableauJobStatusSensor with DAG( dag_id='example_tableau_refresh_workbook', diff --git a/airflow/providers/tableau/provider.yaml b/airflow/providers/tableau/provider.yaml index 2e03c45cfc47f..59801fc6786a1 100644 --- a/airflow/providers/tableau/provider.yaml +++ b/airflow/providers/tableau/provider.yaml @@ -54,6 +54,7 @@ sensors: - integration-name: Tableau python-modules: - airflow.providers.tableau.sensors.tableau_job_status + - airflow.providers.tableau.sensors.tableau hooks: - integration-name: Tableau diff --git a/airflow/providers/tableau/sensors/tableau.py b/airflow/providers/tableau/sensors/tableau.py new file mode 100644 index 0000000000000..7602d28076051 --- /dev/null +++ b/airflow/providers/tableau/sensors/tableau.py @@ -0,0 +1,72 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from typing import TYPE_CHECKING, Optional, Sequence + +from airflow.providers.tableau.hooks.tableau import ( + TableauHook, + TableauJobFailedException, + TableauJobFinishCode, +) +from airflow.sensors.base import BaseSensorOperator + +if TYPE_CHECKING: + from airflow.utils.context import Context + + +class TableauJobStatusSensor(BaseSensorOperator): + """ + Watches the status of a Tableau Server Job. + + .. seealso:: https://tableau.github.io/server-client-python/docs/api-ref#jobs + + :param job_id: Id of the job to watch. + :param site_id: The id of the site where the workbook belongs to. + :param tableau_conn_id: The :ref:`Tableau Connection id ` + containing the credentials to authenticate to the Tableau Server. + """ + + template_fields: Sequence[str] = ('job_id',) + + def __init__( + self, + *, + job_id: str, + site_id: Optional[str] = None, + tableau_conn_id: str = 'tableau_default', + **kwargs, + ) -> None: + super().__init__(**kwargs) + self.tableau_conn_id = tableau_conn_id + self.job_id = job_id + self.site_id = site_id + + def poke(self, context: 'Context') -> bool: + """ + Pokes until the job has successfully finished. + + :param context: The task context during execution. + :return: True if it succeeded and False if not. + :rtype: bool + """ + with TableauHook(self.site_id, self.tableau_conn_id) as tableau_hook: + finish_code = tableau_hook.get_job_status(job_id=self.job_id) + self.log.info('Current finishCode is %s (%s)', finish_code.name, finish_code.value) + + if finish_code in (TableauJobFinishCode.ERROR, TableauJobFinishCode.CANCELED): + raise TableauJobFailedException('The Tableau Refresh Workbook Job failed!') + + return finish_code == TableauJobFinishCode.SUCCESS diff --git a/airflow/providers/tableau/sensors/tableau_job_status.py b/airflow/providers/tableau/sensors/tableau_job_status.py index 7602d28076051..d2c39c3c196b2 100644 --- a/airflow/providers/tableau/sensors/tableau_job_status.py +++ b/airflow/providers/tableau/sensors/tableau_job_status.py @@ -14,59 +14,15 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -from typing import TYPE_CHECKING, Optional, Sequence -from airflow.providers.tableau.hooks.tableau import ( - TableauHook, - TableauJobFailedException, - TableauJobFinishCode, -) -from airflow.sensors.base import BaseSensorOperator - -if TYPE_CHECKING: - from airflow.utils.context import Context - - -class TableauJobStatusSensor(BaseSensorOperator): - """ - Watches the status of a Tableau Server Job. - - .. seealso:: https://tableau.github.io/server-client-python/docs/api-ref#jobs +"""This module is deprecated. Please use :mod:`airflow.providers.tableau.sensors.tableau`.""" - :param job_id: Id of the job to watch. - :param site_id: The id of the site where the workbook belongs to. - :param tableau_conn_id: The :ref:`Tableau Connection id ` - containing the credentials to authenticate to the Tableau Server. - """ +import warnings - template_fields: Sequence[str] = ('job_id',) +from airflow.providers.tableau.sensors.tableau import TableauJobStatusSensor # noqa - def __init__( - self, - *, - job_id: str, - site_id: Optional[str] = None, - tableau_conn_id: str = 'tableau_default', - **kwargs, - ) -> None: - super().__init__(**kwargs) - self.tableau_conn_id = tableau_conn_id - self.job_id = job_id - self.site_id = site_id - - def poke(self, context: 'Context') -> bool: - """ - Pokes until the job has successfully finished. - - :param context: The task context during execution. - :return: True if it succeeded and False if not. - :rtype: bool - """ - with TableauHook(self.site_id, self.tableau_conn_id) as tableau_hook: - finish_code = tableau_hook.get_job_status(job_id=self.job_id) - self.log.info('Current finishCode is %s (%s)', finish_code.name, finish_code.value) - - if finish_code in (TableauJobFinishCode.ERROR, TableauJobFinishCode.CANCELED): - raise TableauJobFailedException('The Tableau Refresh Workbook Job failed!') - - return finish_code == TableauJobFinishCode.SUCCESS +warnings.warn( + "This module is deprecated. Please use `airflow.providers.tableau.sensors.tableau`.", + DeprecationWarning, + stacklevel=2, +) diff --git a/dev/provider_packages/prepare_provider_packages.py b/dev/provider_packages/prepare_provider_packages.py index 319f4447fd14c..9f7b1b5f56b19 100755 --- a/dev/provider_packages/prepare_provider_packages.py +++ b/dev/provider_packages/prepare_provider_packages.py @@ -2233,6 +2233,7 @@ def summarise_total_vs_bad_and_warnings(total: int, bad: int, warns: List[warnin '`airflow.providers.amazon.aws.operators.redshift_cluster` as appropriate.', 'This module is deprecated. Please use `airflow.providers.amazon.aws.sensors.redshift_cluster`.', "This module is deprecated. Please use airflow.providers.amazon.aws.transfers.sql_to_s3`.", + "This module is deprecated. Please use `airflow.providers.tableau.sensors.tableau`.", } diff --git a/tests/providers/tableau/sensors/test_tableau_job_status.py b/tests/providers/tableau/sensors/test_tableau.py similarity index 91% rename from tests/providers/tableau/sensors/test_tableau_job_status.py rename to tests/providers/tableau/sensors/test_tableau.py index 022a836ebce25..7a57462dc80f3 100644 --- a/tests/providers/tableau/sensors/test_tableau_job_status.py +++ b/tests/providers/tableau/sensors/test_tableau.py @@ -21,7 +21,7 @@ import pytest from parameterized import parameterized -from airflow.providers.tableau.sensors.tableau_job_status import ( +from airflow.providers.tableau.sensors.tableau import ( TableauJobFailedException, TableauJobFinishCode, TableauJobStatusSensor, @@ -36,7 +36,7 @@ class TestTableauJobStatusSensor(unittest.TestCase): def setUp(self): self.kwargs = {'job_id': 'job_2', 'site_id': 'test_site', 'task_id': 'task', 'dag': None} - @patch('airflow.providers.tableau.sensors.tableau_job_status.TableauHook') + @patch('airflow.providers.tableau.sensors.tableau.TableauHook') def test_poke(self, mock_tableau_hook): """ Test poke @@ -51,7 +51,7 @@ def test_poke(self, mock_tableau_hook): mock_tableau_hook.get_job_status.assert_called_once_with(job_id=sensor.job_id) @parameterized.expand([(TableauJobFinishCode.ERROR,), (TableauJobFinishCode.CANCELED,)]) - @patch('airflow.providers.tableau.sensors.tableau_job_status.TableauHook') + @patch('airflow.providers.tableau.sensors.tableau.TableauHook') def test_poke_failed(self, finish_code, mock_tableau_hook): """ Test poke failed