Add Tableau provider separate from Salesforce Provider (#14030)

Closes #13614

CONTRIBUTING.rst

@@ -667,6 +667,7 @@ microsoft.mssql            odbc
 mysql                      amazon,presto,vertica
 opsgenie                   http
 postgres                   amazon
+salesforce                 tableau
 sftp                       ssh
 slack                      http
 snowflake                  slack

airflow/providers/dependencies.json

@@ -70,6 +70,9 @@
   "postgres": [
     "amazon"
   ],
+  "salesforce": [
+    "tableau"
+  ],
   "sftp": [
     "ssh"
   ],

airflow/providers/salesforce/CHANGELOG.rst

@@ -19,11 +19,22 @@
 Changelog
 ---------
 
+1.0.2
+.....
+
+Tableau provider moved to separate 'tableau' provider
+
+Things done:
+
+    - Tableau classes imports classes from 'tableau' provider with deprecation warning
+
+
 1.0.1
 .....
 
 Updated documentation and readme files.
 
+
 1.0.0
 .....
 

airflow/providers/salesforce/hooks/tableau.py

@@ -14,102 +14,14 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from enum import Enum
-from typing import Any, Optional
 
-from tableauserverclient import Pager, PersonalAccessTokenAuth, Server, TableauAuth
-from tableauserverclient.server import Auth
+import warnings
 
-from airflow.hooks.base import BaseHook
+# pylint: disable=unused-import
+from airflow.providers.tableau.hooks.tableau import TableauHook, TableauJobFinishCode  # noqa
 
-
-class TableauJobFinishCode(Enum):
-    """
-    The finish code indicates the status of the job.
-
-    .. seealso:: https://help.tableau.com/current/api/rest_api/en-us/REST/rest_api_ref.htm#query_job
-
-    """
-
-    PENDING = -1
-    SUCCESS = 0
-    ERROR = 1
-    CANCELED = 2
-
-
-class TableauHook(BaseHook):
-    """
-    Connects to the Tableau Server Instance and allows to communicate with it.
-
-    .. seealso:: https://tableau.github.io/server-client-python/docs/
-
-    :param site_id: The id of the site where the workbook belongs to.
-        It will connect to the default site if you don't provide an id.
-    :type site_id: Optional[str]
-    :param tableau_conn_id: The Tableau Connection id containing the credentials
-        to authenticate to the Tableau Server.
-    :type tableau_conn_id: str
-    """
-
-    conn_name_attr = 'tableau_conn_id'
-    default_conn_name = 'tableau_default'
-    conn_type = 'tableau'
-    hook_name = 'Tableau'
-
-    def __init__(self, site_id: Optional[str] = None, tableau_conn_id: str = default_conn_name) -> None:
-        super().__init__()
-        self.tableau_conn_id = tableau_conn_id
-        self.conn = self.get_connection(self.tableau_conn_id)
-        self.site_id = site_id or self.conn.extra_dejson.get('site_id', '')
-        self.server = Server(self.conn.host, use_server_version=True)
-        self.tableau_conn = None
-
-    def __enter__(self):
-        if not self.tableau_conn:
-            self.tableau_conn = self.get_conn()
-        return self
-
-    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
-        self.server.auth.sign_out()
-
-    def get_conn(self) -> Auth.contextmgr:
-        """
-        Signs in to the Tableau Server and automatically signs out if used as ContextManager.
-
-        :return: an authorized Tableau Server Context Manager object.
-        :rtype: tableauserverclient.server.Auth.contextmgr
-        """
-        if self.conn.login and self.conn.password:
-            return self._auth_via_password()
-        if 'token_name' in self.conn.extra_dejson and 'personal_access_token' in self.conn.extra_dejson:
-            return self._auth_via_token()
-        raise NotImplementedError('No Authentication method found for given Credentials!')
-
-    def _auth_via_password(self) -> Auth.contextmgr:
-        tableau_auth = TableauAuth(
-            username=self.conn.login, password=self.conn.password, site_id=self.site_id
-        )
-        return self.server.auth.sign_in(tableau_auth)
-
-    def _auth_via_token(self) -> Auth.contextmgr:
-        tableau_auth = PersonalAccessTokenAuth(
-            token_name=self.conn.extra_dejson['token_name'],
-            personal_access_token=self.conn.extra_dejson['personal_access_token'],
-            site_id=self.site_id,
-        )
-        return self.server.auth.sign_in_with_personal_access_token(tableau_auth)
-
-    def get_all(self, resource_name: str) -> Pager:
-        """
-        Get all items of the given resource.
-
-        .. seealso:: https://tableau.github.io/server-client-python/docs/page-through-results
-
-        :param resource_name: The name of the resource to paginate.
-            For example: jobs or workbooks
-        :type resource_name: str
-        :return: all items by returning a Pager.
-        :rtype: tableauserverclient.Pager
-        """
-        resource = getattr(self.server, resource_name)
-        return Pager(resource.get)
+warnings.warn(
+    "This module is deprecated. Please use `airflow.providers.tableau.hooks.tableau`.",
+    DeprecationWarning,
+    stacklevel=2,
+)

airflow/providers/salesforce/operators/tableau_refresh_workbook.py

@@ -14,84 +14,16 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from typing import Optional
 
-from tableauserverclient import WorkbookItem
+import warnings
 
-from airflow.exceptions import AirflowException
-from airflow.models import BaseOperator
-from airflow.providers.salesforce.hooks.tableau import TableauHook
-from airflow.utils.decorators import apply_defaults
+# pylint: disable=unused-import
+from airflow.providers.tableau.operators.tableau_refresh_workbook import (  # noqa
+    TableauRefreshWorkbookOperator,
+)
 
-
-class TableauRefreshWorkbookOperator(BaseOperator):
-    """
-    Refreshes a Tableau Workbook/Extract
-
-    .. seealso:: https://tableau.github.io/server-client-python/docs/api-ref#workbooks
-
-    :param workbook_name: The name of the workbook to refresh.
-    :type workbook_name: str
-    :param site_id: The id of the site where the workbook belongs to.
-    :type site_id: Optional[str]
-    :param blocking: By default the extract refresh will be blocking means it will wait until it has finished.
-    :type blocking: bool
-    :param tableau_conn_id: The Tableau Connection id containing the credentials
-        to authenticate to the Tableau Server.
-    :type tableau_conn_id: str
-    """
-
-    @apply_defaults
-    def __init__(
-        self,
-        *,
-        workbook_name: str,
-        site_id: Optional[str] = None,
-        blocking: bool = True,
-        tableau_conn_id: str = 'tableau_default',
-        **kwargs,
-    ) -> None:
-        super().__init__(**kwargs)
-        self.workbook_name = workbook_name
-        self.site_id = site_id
-        self.blocking = blocking
-        self.tableau_conn_id = tableau_conn_id
-
-    def execute(self, context: dict) -> str:
-        """
-        Executes the Tableau Extract Refresh and pushes the job id to xcom.
-
-        :param context: The task context during execution.
-        :type context: dict
-        :return: the id of the job that executes the extract refresh
-        :rtype: str
-        """
-        with TableauHook(self.site_id, self.tableau_conn_id) as tableau_hook:
-            workbook = self._get_workbook_by_name(tableau_hook)
-
-            job_id = self._refresh_workbook(tableau_hook, workbook.id)
-            if self.blocking:
-                from airflow.providers.salesforce.sensors.tableau_job_status import TableauJobStatusSensor
-
-                TableauJobStatusSensor(
-                    job_id=job_id,
-                    site_id=self.site_id,
-                    tableau_conn_id=self.tableau_conn_id,
-                    task_id='wait_until_succeeded',
-                    dag=None,
-                ).execute(context={})
-                self.log.info('Workbook %s has been successfully refreshed.', self.workbook_name)
-            return job_id
-
-    def _get_workbook_by_name(self, tableau_hook: TableauHook) -> WorkbookItem:
-        for workbook in tableau_hook.get_all(resource_name='workbooks'):
-            if workbook.name == self.workbook_name:
-                self.log.info('Found matching workbook with id %s', workbook.id)
-                return workbook
-
-        raise AirflowException(f'Workbook {self.workbook_name} not found!')
-
-    def _refresh_workbook(self, tableau_hook: TableauHook, workbook_id: str) -> str:
-        job = tableau_hook.server.workbooks.refresh(workbook_id)
-        self.log.info('Refreshing Workbook %s...', self.workbook_name)
-        return job.id
+warnings.warn(
+    "This module is deprecated. Please use `airflow.providers.tableau.operators.tableau_refresh_workbook`.",
+    DeprecationWarning,
+    stacklevel=2,
+)

airflow/providers/salesforce/provider.yaml

@@ -22,6 +22,7 @@ description: |
     `Salesforce <https://www.salesforce.com/>`__
 
 versions:
+  - 1.0.2
   - 1.0.1
   - 1.0.0
 
@@ -42,10 +43,12 @@ sensors:
       - airflow.providers.salesforce.sensors.tableau_job_status
 
 hooks:
+  - integration-name: Tableau
+    python-modules:
+      - airflow.providers.salesforce.hooks.tableau
   - integration-name: Salesforce
     python-modules:
       - airflow.providers.salesforce.hooks.salesforce
-      - airflow.providers.salesforce.hooks.tableau
 
 hook-class-names:
   - airflow.providers.salesforce.hooks.tableau.TableauHook

airflow/providers/salesforce/sensors/tableau_job_status.py

@@ -14,63 +14,17 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from typing import Optional
 
-from airflow.exceptions import AirflowException
-from airflow.providers.salesforce.hooks.tableau import TableauHook, TableauJobFinishCode
-from airflow.sensors.base import BaseSensorOperator
-from airflow.utils.decorators import apply_defaults
+import warnings
 
+# pylint: disable=unused-import
+from airflow.providers.tableau.sensors.tableau_job_status import (  # noqa
+    TableauJobFailedException,
+    TableauJobStatusSensor,
+)
 
-class TableauJobFailedException(AirflowException):
-    """An exception that indicates that a Job failed to complete."""
-
-
-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: The job to watch.
-    :type job_id: str
-    :param site_id: The id of the site where the workbook belongs to.
-    :type site_id: Optional[str]
-    :param tableau_conn_id: The Tableau Connection id containing the credentials
-        to authenticate to the Tableau Server.
-    :type tableau_conn_id: str
-    """
-
-    template_fields = ('job_id',)
-
-    @apply_defaults
-    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: dict) -> bool:
-        """
-        Pokes until the job has successfully finished.
-
-        :param context: The task context during execution.
-        :type context: dict
-        :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 = TableauJobFinishCode(
-                int(tableau_hook.server.jobs.get_by_id(self.job_id).finish_code)
-            )
-            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_job_status`.",
+    DeprecationWarning,
+    stacklevel=2,
+)

airflow/providers/tableau/CHANGELOG.rst

@@ -0,0 +1,25 @@
+ .. 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.
+
+
+Changelog
+---------
+
+1.0.0
+.....
+
+Initial version of the provider.

airflow/providers/tableau/__init__.py

@@ -0,0 +1,17 @@
+#
+# 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.