Tableau Refresh Operators

TableauRefreshDataSourceOperator and TableauRefreshWorkBookOperator refresh Tableau data sources or workbooks by triggering async refresh jobs and polling the Tableau server until they complete.

Previously these operators subclassed Airflow's BaseOperator. They have been rewritten as plain Python classes with an execute() method -- no apache-airflow install is required. The public constructor signature and behavior remain unchanged apart from no longer accepting task_id (which was an Airflow concept).

from brickflow_plugins import (
    TableauRefreshDataSourceOperator,
    TableauRefreshWorkBookOperator,
)

TableauRefreshDataSourceOperator(
    server="https://tableau.example.com",
    username="me",
    password="pw",
    site="my_site",
    project="my_project",
    data_sources=["ds_a", "ds_b"],
).execute()

Requires the tableauserverclient library to be installed on the cluster (via PypiTaskLibrary("tableauserverclient==0.25") or the brickflow[tableau] extra during local development).

API Reference

Tableau Refresh Operators.

Native brickflow operators that refresh Tableau data sources or workbooks by triggering async refresh jobs and polling the Tableau server until they complete. No Airflow dependency -- plain Python classes with an execute() method.

Attributes

Classes

brickflow_plugins.operators.tableau_refresh_operator.TableauRefreshABCOperator(server: str, username: str, password: str, site: str, project: Optional[str] = None, parent_project: Optional[str] = None, version: str = '3.14', max_async_workers: int = 5, polling_required: bool = True, polling_interval: int = 30, polling_timeout: int = 600, fail_operator: bool = True)

Bases: ABC

Abstract base class that implements generic functionality for Tableau refresh operators. No Airflow inheritance -- plain Python.

Source code in brickflow_plugins/operators/tableau_refresh_operator.py

def __init__(
    self,
    server: str,
    username: str,
    password: str,
    site: str,
    project: Optional[str] = None,
    parent_project: Optional[str] = None,
    version: str = "3.14",
    max_async_workers: int = 5,
    polling_required: bool = True,
    polling_interval: int = 30,
    polling_timeout: int = 600,
    fail_operator: bool = True,
) -> None:
    _ensure_tableau()
    self._logger = log

    self.wrapper_options = {
        "server": server,
        "username": username,
        "password": password,
        "site": site,
        "project": project,
        "parent_project": parent_project,
        "version": version,
        "max_async_workers": max_async_workers,
        "polling_required": polling_required,
        "polling_interval": polling_interval,
        "polling_timeout": polling_timeout,
    }

    self._logger.info(f"Tableau wrapper options:{self.wrapper_options}")

    self.tableau_wrapper: Optional[TableauWrapper] = None
    self.fail_operator = fail_operator

Attributes

fail_operator = fail_operator instance-attribute

tableau_wrapper: Optional[TableauWrapper] = None instance-attribute

wrapper_options = {'server': server, 'username': username, 'password': password, 'site': site, 'project': project, 'parent_project': parent_project, 'version': version, 'max_async_workers': max_async_workers, 'polling_required': polling_required, 'polling_interval': polling_interval, 'polling_timeout': polling_timeout} instance-attribute

Functions

execute() abstractmethod

Source code in brickflow_plugins/operators/tableau_refresh_operator.py

@abstractmethod
def execute(self):
    raise NotImplementedError

brickflow_plugins.operators.tableau_refresh_operator.TableauRefreshDataSourceOperator(data_sources: list, skip: bool = False, **kwargs)

Bases: TableauRefreshABCOperator

Refresh a list of Tableau data sources.

Example

::

TableauRefreshDataSourceOperator(
    server="https://tableau.example.com",
    username="me",
    password="pw",
    site="my_site",
    project="my_project",
    data_sources=["ds_a", "ds_b"],
).execute()

Source code in brickflow_plugins/operators/tableau_refresh_operator.py

def __init__(
    self,
    data_sources: list,
    skip: bool = False,
    **kwargs,
) -> None:
    super().__init__(**kwargs)
    self.data_sources = data_sources
    self._skip = skip

Attributes

data_sources = data_sources instance-attribute

Functions

execute()

Refresh data sources in Tableau.

Source code in brickflow_plugins/operators/tableau_refresh_operator.py

def execute(self):
    """Refresh data sources in Tableau."""
    if not self._skip:
        self.tableau_wrapper = TableauWrapper(**self.wrapper_options)
        results = self.tableau_wrapper.refresh_datasources(
            data_sources=self.data_sources
        )
        self._analyze_refresh_result(results)
    else:
        self._logger.info("Skipping task execution...")

brickflow_plugins.operators.tableau_refresh_operator.TableauRefreshEmptyException

Bases: Exception

brickflow_plugins.operators.tableau_refresh_operator.TableauRefreshException

Bases: Exception

brickflow_plugins.operators.tableau_refresh_operator.TableauRefreshWorkBookOperator(workbooks: list, skip: bool = False, **kwargs)

Bases: TableauRefreshABCOperator

Refresh a list of Tableau workbooks.

Example

::

TableauRefreshWorkBookOperator(
    server="https://tableau.example.com",
    username="me",
    password="pw",
    site="my_site",
    project="my_project",
    workbooks=["wb_a", "wb_b"],
).execute()

Source code in brickflow_plugins/operators/tableau_refresh_operator.py

def __init__(
    self,
    workbooks: list,
    skip: bool = False,
    **kwargs,
) -> None:
    super().__init__(**kwargs)
    self.workbooks = workbooks
    self._skip = skip

Attributes

workbooks = workbooks instance-attribute

Functions

execute()

Refresh workbooks in Tableau.

Source code in brickflow_plugins/operators/tableau_refresh_operator.py

def execute(self):
    """Refresh workbooks in Tableau."""
    if not self._skip:
        self.tableau_wrapper = TableauWrapper(**self.wrapper_options)
        results = self.tableau_wrapper.refresh_workbooks(work_books=self.workbooks)
        self._analyze_refresh_result(results)
    else:
        self._logger.info("Skipping task execution...")

brickflow_plugins.operators.tableau_refresh_operator.TableauWrapper(server: str, username: str, password: str, site: str, project: Optional[str] = None, parent_project: Optional[str] = None, version: str = '3.14', max_async_workers: int = 5, polling_required: bool = True, polling_interval: int = 30, polling_timeout: int = 600)

Facilitates interaction with a Tableau server for the purpose of refreshing data sources or workbooks. Refresh is triggered asynchronously, and the Tableau server is polled until every job is finished or the polling_timeout is reached.

Parameters

server : str Tableau server address, e.g. https://tableau-server.com. username : str Log in username. password : str Log in password. site : str Tableau site. project : str Tableau project. parent_project : str Name of the parent Tableau project. Use "/" for the site root. version : str Tableau server API version. max_async_workers : int Maximum number of asynchronous tasks that will trigger jobs and wait for completion. polling_required : bool Wait for job completion to proceed, otherwise trigger the job and proceed without waiting. polling_interval : int Polling interval for the job status updates (seconds). polling_timeout : int Stop polling if the job was not completed within the specified interval (seconds).

Source code in brickflow_plugins/operators/tableau_refresh_operator.py

def __init__(
    self,
    server: str,
    username: str,
    password: str,
    site: str,
    project: Optional[str] = None,
    parent_project: Optional[str] = None,
    version: str = "3.14",
    max_async_workers: int = 5,
    polling_required: bool = True,
    polling_interval: int = 30,
    polling_timeout: int = 600,
) -> None:
    """
    Parameters
    ----------
    server : str
        Tableau server address, e.g. ``https://tableau-server.com``.
    username : str
        Log in username.
    password : str
        Log in password.
    site : str
        Tableau site.
    project : str
        Tableau project.
    parent_project : str
        Name of the parent Tableau project. Use ``"/"`` for the site root.
    version : str
        Tableau server API version.
    max_async_workers : int
        Maximum number of asynchronous tasks that will trigger jobs and
        wait for completion.
    polling_required : bool
        Wait for job completion to proceed, otherwise trigger the job and
        proceed without waiting.
    polling_interval : int
        Polling interval for the job status updates (seconds).
    polling_timeout : int
        Stop polling if the job was not completed within the specified
        interval (seconds).
    """
    _ensure_tableau()
    self.server = server
    self.version = version
    self.username = username
    self.password = password
    self.site = site
    self.project = project
    self.parent_project = parent_project

    self.max_async_workers = max_async_workers
    self.polling_required = polling_required
    self.polling_interval = polling_interval
    self.polling_timeout = polling_timeout

    self._logger = log
    self._ip = None

Attributes

max_async_workers = max_async_workers instance-attribute

parent_project = parent_project instance-attribute

password = password instance-attribute

polling_interval = polling_interval instance-attribute

polling_required = polling_required instance-attribute

polling_timeout = polling_timeout instance-attribute

project = project instance-attribute

server = server instance-attribute

site = site instance-attribute

username = username instance-attribute

version = version instance-attribute

Classes

MultipleWorkingProjectsException()

Bases: Exception

Source code in brickflow_plugins/operators/tableau_refresh_operator.py

def __init__(self):
    self.message = (
        "Multiple projects with the same name exist on the server! Set "
        "'parent_project' parameter!"
    )
    super().__init__(self.message)

Attributes

message = "Multiple projects with the same name exist on the server! Set 'parent_project' parameter!" instance-attribute

UnidentifiedWorkingProjectException()

Bases: Exception

Source code in brickflow_plugins/operators/tableau_refresh_operator.py

def __init__(self):
    self.message = "Could not identify working project, check that the spelling is correct!"
    super().__init__(self.message)

Attributes

message = 'Could not identify working project, check that the spelling is correct!' instance-attribute

Functions

refresh_datasources(data_sources: list) -> list

Asynchronously refresh a list of Tableau data sources.

Source code in brickflow_plugins/operators/tableau_refresh_operator.py

def refresh_datasources(self, data_sources: list) -> list:
    """Asynchronously refresh a list of Tableau data sources."""
    with self._authenticate():
        # Only refresh selected data sources
        lim_ds = self._filter_datasources(data_sources=data_sources)

        # Start async execution and collect results
        with concurrent.futures.ThreadPoolExecutor(
            max_workers=self.max_async_workers
        ) as executor:
            executor_results = executor.map(self._refresh_datasource, lim_ds)

            results = [result[1] for result in enumerate(executor_results)]

        return results

refresh_workbooks(work_books: list) -> list

Asynchronously refresh a list of Tableau workbooks.

Source code in brickflow_plugins/operators/tableau_refresh_operator.py

def refresh_workbooks(self, work_books: list) -> list:
    """Asynchronously refresh a list of Tableau workbooks."""
    with self._authenticate():
        # Only refresh selected workbooks
        lim_wb = self._filter_workbooks(work_books=work_books)

        # Start async execution and collect results
        with concurrent.futures.ThreadPoolExecutor(
            max_workers=self.max_async_workers
        ) as executor:
            executor_results = executor.map(self._refresh_workbook, lim_wb)

            results = [result[1] for result in enumerate(executor_results)]

        return results