Merge pull request #1 from tableau/development

Add support for materializeViews as schedule and task type (tableau#542)

Author: guodah

tableauserverclient/models/schedule_item.py

@@ -11,6 +11,7 @@ class Type:
         Extract = "Extract"
         Flow = "Flow"
         Subscription = "Subscription"
+        MaterializeViews = "MaterializeViews"
 
     class ExecutionOrder:
         Parallel = "Parallel"
@@ -199,7 +200,7 @@ def _parse_interval_item(parsed_response, frequency, ns):
             # We use fractional hours for the two minute-based intervals.
             # Need to convert to hours from minutes here
             if interval_occurrence == IntervalItem.Occurrence.Minutes:
-                interval_value = float(interval_value / 60)
+                interval_value = float(interval_value) / 60
 
             return HourlyInterval(start_time, end_time, interval_value)
 

tableauserverclient/models/task_item.py

@@ -1,39 +1,52 @@
 import xml.etree.ElementTree as ET
 from .target import Target
+from .schedule_item import ScheduleItem
+from ..datetime_helpers import parse_datetime
 
 
 class TaskItem(object):
-    def __init__(self, id_, task_type, priority, consecutive_failed_count=0, schedule_id=None, target=None):
+    class Type:
+        ExtractRefresh = "extractRefresh"
+        MaterializeViews = "materializeViews"
+
+    def __init__(self, id_, task_type, priority, consecutive_failed_count=0, schedule_id=None,
+                 schedule_item=None, last_run_at=None, target=None):
         self.id = id_
         self.task_type = task_type
         self.priority = priority
         self.consecutive_failed_count = consecutive_failed_count
         self.schedule_id = schedule_id
+        self.schedule_item = schedule_item
+        self.last_run_at = last_run_at
         self.target = target
 
     def __repr__(self):
         return "<Task#{id} {task_type} pri({priority}) failed({consecutive_failed_count}) schedule_id({" \
                "schedule_id}) target({target})>".format(**self.__dict__)
 
     @classmethod
-    def from_response(cls, xml, ns):
+    def from_response(cls, xml, ns, task_type=Type.ExtractRefresh):
         parsed_response = ET.fromstring(xml)
         all_tasks_xml = parsed_response.findall(
-            './/t:task/t:extractRefresh', namespaces=ns)
+            './/t:task/t:{}'.format(task_type), namespaces=ns)
 
         all_tasks = (TaskItem._parse_element(x, ns) for x in all_tasks_xml)
 
         return list(all_tasks)
 
     @classmethod
     def _parse_element(cls, element, ns):
-        schedule = None
+        schedule_id = None
+        schedule_item = None
         target = None
-        schedule_element = element.find('.//t:schedule', namespaces=ns)
+        last_run_at = None
         workbook_element = element.find('.//t:workbook', namespaces=ns)
         datasource_element = element.find('.//t:datasource', namespaces=ns)
-        if schedule_element is not None:
-            schedule = schedule_element.get('id', None)
+        last_run_at_element = element.find('.//t:lastRunAt', namespaces=ns)
+
+        schedule_item_list = ScheduleItem.from_element(element, ns)
+        if len(schedule_item_list) >= 1:
+            schedule_item = schedule_item_list[0]
 
         # according to the Tableau Server REST API documentation,
         # there should be only one of workbook or datasource
@@ -43,9 +56,12 @@ def _parse_element(cls, element, ns):
         if datasource_element is not None:
             datasource_id = datasource_element.get('id', None)
             target = Target(datasource_id, "datasource")
+        if last_run_at_element is not None:
+            last_run_at = parse_datetime(last_run_at_element.text)
 
         task_type = element.get('type', None)
         priority = int(element.get('priority', -1))
         consecutive_failed_count = int(element.get('consecutiveFailedCount', 0))
         id_ = element.get('id', None)
-        return cls(id_, task_type, priority, consecutive_failed_count, schedule, target)
+        return cls(id_, task_type, priority, consecutive_failed_count, schedule_item.id,
+                   schedule_item, last_run_at, target)

tableauserverclient/server/endpoint/schedules_endpoint.py

@@ -1,6 +1,6 @@
 from .endpoint import Endpoint, api
 from .exceptions import MissingRequiredFieldError
-from .. import RequestFactory, PaginationItem, ScheduleItem, WorkbookItem, DatasourceItem
+from .. import RequestFactory, PaginationItem, ScheduleItem, WorkbookItem, DatasourceItem, TaskItem
 import logging
 import copy
 from collections import namedtuple
@@ -68,12 +68,13 @@ def create(self, schedule_item):
         return new_schedule
 
     @api(version="2.8")
-    def add_to_schedule(self, schedule_id, workbook=None, datasource=None):
+    def add_to_schedule(self, schedule_id, workbook=None, datasource=None,
+                        task_type=TaskItem.Type.ExtractRefresh):
 
         def add_to(resource, type_, req_factory):
             id_ = resource.id
             url = "{0}/{1}/{2}s".format(self.siteurl, schedule_id, type_)
-            add_req = req_factory(id_)
+            add_req = req_factory(id_, task_type=task_type)
             response = self.put_request(url, add_req)
             if response.status_code < 200 or response.status_code >= 300:
                 return AddResponse(result=False,

tableauserverclient/server/endpoint/tasks_endpoint.py

@@ -9,26 +9,44 @@
 class Tasks(Endpoint):
     @property
     def baseurl(self):
-        return "{0}/sites/{1}/tasks/extractRefreshes".format(self.parent_srv.baseurl,
-                                                             self.parent_srv.site_id)
+        return "{0}/sites/{1}/tasks".format(self.parent_srv.baseurl,
+                                            self.parent_srv.site_id)
+
+    def __normalize_task_type(self, task_type):
+        """
+            The word for extract refresh used in API URL is "extractRefreshes".
+            It is different than the tag "extractRefresh" used in the request body.
+        """
+        if task_type == TaskItem.Type.ExtractRefresh:
+            return '{}es'.format(task_type)
+        else:
+            return task_type
 
     @api(version='2.6')
-    def get(self, req_options=None):
-        logger.info('Querying all tasks for the site')
-        url = self.baseurl
+    def get(self, req_options=None, task_type=TaskItem.Type.ExtractRefresh):
+        if task_type == TaskItem.Type.MaterializeViews:
+            self.parent_srv.assert_at_least_version("3.8")
+
+        logger.info('Querying all {} tasks for the site'.format(task_type))
+
+        url = "{0}/{1}".format(self.baseurl, self.__normalize_task_type(task_type))
         server_response = self.get_request(url, req_options)
 
-        pagination_item = PaginationItem.from_response(server_response.content, self.parent_srv.namespace)
-        all_extract_tasks = TaskItem.from_response(server_response.content, self.parent_srv.namespace)
-        return all_extract_tasks, pagination_item
+        pagination_item = PaginationItem.from_response(server_response.content,
+                                                       self.parent_srv.namespace)
+        all_tasks = TaskItem.from_response(server_response.content,
+                                           self.parent_srv.namespace,
+                                           task_type)
+        return all_tasks, pagination_item
 
     @api(version='2.6')
     def get_by_id(self, task_id):
         if not task_id:
             error = "No Task ID provided"
             raise ValueError(error)
         logger.info("Querying a single task by id ({})".format(task_id))
-        url = "{}/{}".format(self.baseurl, task_id)
+        url = "{}/{}/{}".format(self.baseurl,
+                                self.__normalize_task_type(TaskItem.Type.ExtractRefresh), task_id)
         server_response = self.get_request(url)
         return TaskItem.from_response(server_response.content, self.parent_srv.namespace)[0]
 
@@ -38,17 +56,22 @@ def run(self, task_item):
             error = "User item missing ID."
             raise MissingRequiredFieldError(error)
 
-        url = "{0}/{1}/runNow".format(self.baseurl, task_item.id)
+        url = "{0}/{1}/{2}/runNow".format(self.baseurl,
+                                          self.__normalize_task_type(TaskItem.Type.ExtractRefresh), task_item.id)
         run_req = RequestFactory.Task.run_req(task_item)
         server_response = self.post_request(url, run_req)
         return server_response.content
 
     # Delete 1 task by id
     @api(version="3.6")
-    def delete(self, task_id):
+    def delete(self, task_id, task_type=TaskItem.Type.ExtractRefresh):
+        if task_type == TaskItem.Type.MaterializeViews:
+            self.parent_srv.assert_at_least_version("3.8")
+
         if not task_id:
             error = "No Task ID provided"
             raise ValueError(error)
-        url = "{0}/{1}".format(self.baseurl, task_id)
+        url = "{0}/{1}/{2}".format(self.baseurl,
+                                   self.__normalize_task_type(task_type), task_id)
         self.delete_request(url)
         logger.info('Deleted single task (ID: {0})'.format(task_id))

tableauserverclient/server/request_factory.py

@@ -5,7 +5,7 @@
 from requests.packages.urllib3.fields import RequestField
 from requests.packages.urllib3.filepost import encode_multipart_formdata
 
-from ..models import UserItem, GroupItem, PermissionsRule
+from ..models import TaskItem, UserItem, GroupItem, PermissionsRule
 
 
 def _add_multipart(parts):
@@ -24,6 +24,7 @@ def wrapper(self, *args, **kwargs):
         xml_request = ET.Element('tsRequest')
         func(self, xml_request, *args, **kwargs)
         return ET.tostring(xml_request)
+
     return wrapper
 
 
@@ -311,28 +312,28 @@ def update_req(self, schedule_item):
                 single_interval_element.attrib[expression] = value
         return ET.tostring(xml_request)
 
-    def _add_to_req(self, id_, type_):
+    def _add_to_req(self, id_, target_type, task_type=TaskItem.Type.ExtractRefresh):
         """
         <task>
-          <extractRefresh>
+          <target_type>
             <workbook/datasource id="..."/>
-          </extractRefresh>
+          </target_type>
         </task>
 
         """
         xml_request = ET.Element('tsRequest')
         task_element = ET.SubElement(xml_request, 'task')
-        refresh = ET.SubElement(task_element, 'extractRefresh')
-        workbook = ET.SubElement(refresh, type_)
+        task = ET.SubElement(task_element, task_type)
+        workbook = ET.SubElement(task, target_type)
         workbook.attrib['id'] = id_
 
         return ET.tostring(xml_request)
 
-    def add_workbook_req(self, id_):
-        return self._add_to_req(id_, "workbook")
+    def add_workbook_req(self, id_, task_type=TaskItem.Type.ExtractRefresh):
+        return self._add_to_req(id_, "workbook", task_type)
 
-    def add_datasource_req(self, id_):
-        return self._add_to_req(id_, "datasource")
+    def add_datasource_req(self, id_, task_type=TaskItem.Type.ExtractRefresh):
+        return self._add_to_req(id_, "datasource", task_type)
 
 
 class SiteRequest(object):
@@ -479,7 +480,7 @@ def update_req(self, workbook_item):
         if workbook_item.owner_id:
             owner_element = ET.SubElement(workbook_element, 'owner')
             owner_element.attrib['id'] = workbook_item.owner_id
-        if workbook_item.materialized_views_config['materialized_views_enabled']\
+        if workbook_item.materialized_views_config['materialized_views_enabled'] \
                 and workbook_item.materialized_views_config['run_materialization_now']:
             materialized_views_config = workbook_item.materialized_views_config
             materialized_views_element = ET.SubElement(workbook_element, 'materializedViewsEnablementConfig')

test/assets/tasks_run_now_response.xml

@@ -0,0 +1,6 @@
+<?xml version='1.0' encoding='UTF-8'?>
+<tsResponse
+    xmlns="http://tableau.com/api"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://tableau.com/api http://tableau.com/api/ts-api-2.6.xsd">
+    <job id="7b6b59a8-ac3c-4d1d-2e9e-0b5b4ba8a7b6" mode="Asynchronous" type="RefreshExtract" />
+</tsResponse>

test/assets/tasks_with_materializeviews_task.xml

@@ -0,0 +1,18 @@
+<?xml version='1.0' encoding='UTF-8'?>
+<tsResponse xmlns="http://tableau.com/api" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://tableau.com/api http://tableau.com/api/ts-api-3.8.xsd">
+    <tasks>
+        <task>
+            <materializeViews id="2b217acb-194a-4291-ae90-7d5ec360395d" consecutiveFailedCount="0" type="MaterializeViewsTask">
+                <schedule id="b22190b4-6ac2-4eed-9563-4afc03444413" name="Hourly4-Schedule" state="Active" priority="75" createdAt="2019-12-06T03:27:35Z" updatedAt="2019-12-09T20:30:59Z" type="MaterializeViews" frequency="Hourly" nextRunAt="2019-12-09T22:30:00Z">
+                    <frequencyDetails start="02:30:00" end="23:00:00">
+                        <intervals>
+                            <interval hours="2"/>
+                        </intervals>
+                    </frequencyDetails>
+                </schedule>
+                <workbook id="a462c148-fc40-4670-a8e4-39b7f0c58c7f"/>
+                <lastRunAt>2019-12-09T20:45:04Z</lastRunAt>
+            </materializeViews>
+        </task>
+    </tasks>
+</tsResponse>

test/test_task.py

@@ -2,25 +2,30 @@
 import os
 import requests_mock
 import tableauserverclient as TSC
+from tableauserverclient.models.task_item import TaskItem
+from tableauserverclient.datetime_helpers import parse_datetime
 
 TEST_ASSET_DIR = os.path.join(os.path.dirname(__file__), "assets")
 
 GET_XML_NO_WORKBOOK = os.path.join(TEST_ASSET_DIR, "tasks_no_workbook_or_datasource.xml")
 GET_XML_WITH_WORKBOOK = os.path.join(TEST_ASSET_DIR, "tasks_with_workbook.xml")
 GET_XML_WITH_DATASOURCE = os.path.join(TEST_ASSET_DIR, "tasks_with_datasource.xml")
 GET_XML_WITH_WORKBOOK_AND_DATASOURCE = os.path.join(TEST_ASSET_DIR, "tasks_with_workbook_and_datasource.xml")
+GET_XML_MATERIALIZEVIEWS_TASK = os.path.join(TEST_ASSET_DIR, "tasks_with_materializeviews_task.xml")
+GET_XML_RUN_NOW_RESPONSE = os.path.join(TEST_ASSET_DIR, "tasks_run_now_response.xml")
 
 
 class TaskTests(unittest.TestCase):
     def setUp(self):
         self.server = TSC.Server("http://test")
-        self.server.version = '3.6'
+        self.server.version = '3.8'
 
         # Fake Signin
         self.server._site_id = "dad65087-b08b-4603-af4e-2887b8aafc67"
         self.server._auth_token = "j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM"
 
-        self.baseurl = self.server.tasks.baseurl
+        # default task type is extractRefreshes
+        self.baseurl = "{}/{}".format(self.server.tasks.baseurl, "extractRefreshes")
 
     def test_get_tasks_with_no_workbook(self):
         with open(GET_XML_NO_WORKBOOK, "rb") as f:
@@ -84,3 +89,50 @@ def test_delete(self):
 
     def test_delete_missing_id(self):
         self.assertRaises(ValueError, self.server.tasks.delete, '')
+
+    def test_get_materializeviews_tasks(self):
+        with open(GET_XML_MATERIALIZEVIEWS_TASK, "rb") as f:
+            response_xml = f.read().decode("utf-8")
+        with requests_mock.mock() as m:
+            m.get('{}/{}'.format(
+                self.server.tasks.baseurl, TaskItem.Type.MaterializeViews), text=response_xml)
+            all_tasks, pagination_item = self.server.tasks.get(task_type=TaskItem.Type.MaterializeViews)
+
+        task = all_tasks[0]
+        self.assertEqual('a462c148-fc40-4670-a8e4-39b7f0c58c7f', task.target.id)
+        self.assertEqual('workbook', task.target.type)
+        self.assertEqual('b22190b4-6ac2-4eed-9563-4afc03444413', task.schedule_id)
+        self.assertEqual(parse_datetime('2019-12-09T22:30:00Z'), task.schedule_item.next_run_at)
+        self.assertEqual(parse_datetime('2019-12-09T20:45:04Z'), task.last_run_at)
+
+    def test_delete(self):
+        with requests_mock.mock() as m:
+            m.delete('{}/{}/{}'.format(
+                self.server.tasks.baseurl, TaskItem.Type.MaterializeViews,
+                'c9cff7f9-309c-4361-99ff-d4ba8c9f5467'), status_code=204)
+            self.server.tasks.delete('c9cff7f9-309c-4361-99ff-d4ba8c9f5467',
+                                     TaskItem.Type.MaterializeViews)
+
+    def test_get_by_id(self):
+        with open(GET_XML_WITH_WORKBOOK, "rb") as f:
+            response_xml = f.read().decode("utf-8")
+        task_id = 'f84901ac-72ad-4f9b-a87e-7a3500402ad6'
+        with requests_mock.mock() as m:
+            m.get('{}/{}'.format(self.baseurl, task_id), text=response_xml)
+            task = self.server.tasks.get_by_id(task_id)
+
+        self.assertEqual('c7a9327e-1cda-4504-b026-ddb43b976d1d', task.target.id)
+        self.assertEqual('workbook', task.target.type)
+        self.assertEqual('b60b4efd-a6f7-4599-beb3-cb677e7abac1', task.schedule_id)
+
+    def test_run_now(self):
+        task_id = 'f84901ac-72ad-4f9b-a87e-7a3500402ad6'
+        task = TaskItem(task_id, TaskItem.Type.ExtractRefresh, 100)
+        with open(GET_XML_RUN_NOW_RESPONSE, "rb") as f:
+            response_xml = f.read().decode("utf-8")
+        with requests_mock.mock() as m:
+            m.post('{}/{}/runNow'.format(self.baseurl, task_id), text=response_xml)
+            job_response_content = self.server.tasks.run(task).decode("utf-8")
+
+        self.assertTrue('7b6b59a8-ac3c-4d1d-2e9e-0b5b4ba8a7b6' in job_response_content)
+        self.assertTrue('RefreshExtract' in job_response_content)