tableau · jacalata · Apr 5, 2022 · Mar 30, 2022 · Mar 31, 2022 · Mar 31, 2022
diff --git a/samples/set_refresh_schedule.py b/samples/set_refresh_schedule.py
@@ -34,6 +34,7 @@ def usage(args):
     group = parser.add_mutually_exclusive_group(required=True)
     group.add_argument("--workbook", "-w")
     group.add_argument("--datasource", "-d")
+    group.add_argument("--flow", "-f")
     parser.add_argument("schedule")
 
     return parser.parse_args(args)
@@ -61,6 +62,13 @@ def get_datasource_by_name(server, name):
     return datasources.pop()
 
 
+def get_flow_by_name(server, name):
+    request_filter = make_filter(Name=name)
+    flows, _ = server.flows.get(request_filter)
+    assert len(flows) == 1
+    return flows.pop()
+
+
 def get_schedule_by_name(server, name):
     schedules = [x for x in TSC.Pager(server.schedules) if x.name == name]
     assert len(schedules) == 1
@@ -82,8 +90,13 @@ def run(args):
     with server.auth.sign_in(tableau_auth):
         if args.workbook:
             item = get_workbook_by_name(server, args.workbook)
-        else:
+        elif args.datasource:
             item = get_datasource_by_name(server, args.datasource)
+        elif args.flow:
+            item = get_flow_by_name(server, args.flow)
+        else:
+            print("A scheduleable item must be included")
+            return
         schedule = get_schedule_by_name(server, args.schedule)
 
         assign_to_schedule(server, item, schedule)

diff --git a/tableauserverclient/models/task_item.py b/tableauserverclient/models/task_item.py
@@ -9,6 +9,7 @@ class TaskItem(object):
     class Type:
         ExtractRefresh = "extractRefresh"
         DataAcceleration = "dataAcceleration"
+        RunFlow = "runFlow"
 
     # This mapping is used to convert task type returned from server
     _TASK_TYPE_MAPPING = {

diff --git a/tableauserverclient/server/endpoint/datasources_endpoint.py b/tableauserverclient/server/endpoint/datasources_endpoint.py
@@ -460,5 +460,10 @@ def delete_revision(self, datasource_id: str, revision_number: str) -> None:
 
         self.delete_request(url)
         logger.info(
-            "Deleted single datasource revsision (ID: {0}) (Revision: {1})".format(datasource_id, revision_number)
+            "Deleted single datasource revision (ID: {0}) (Revision: {1})".format(datasource_id, revision_number)
         )
+
+    # a convenience method
+    @api(version="2.8")
+    def schedule_extract_refresh(self, schedule_id: int, item: DatasourceItem) -> None:  # actually should return a task
+        return self.parent_srv.schedules.add_to_schedule(schedule_id, datasource=item)
diff --git a/tableauserverclient/server/endpoint/flows_endpoint.py b/tableauserverclient/server/endpoint/flows_endpoint.py
@@ -237,3 +237,8 @@ def add_dqw(self, item: FlowItem, warning: "DQWItem") -> None:
     @api(version="3.5")
     def delete_dqw(self, item: FlowItem) -> None:
         self._data_quality_warnings.clear(item)
+
+    # a convenience method
+    @api(version="3.3")
+    def schedule_flow_run(self, schedule_id: int, item: FlowItem) -> None:  # actually should return a task
+        return self.parent_srv.schedules.add_to_schedule(schedule_id, flow=item)
diff --git a/tableauserverclient/server/endpoint/schedules_endpoint.py b/tableauserverclient/server/endpoint/schedules_endpoint.py
@@ -1,20 +1,20 @@
 import copy
 import logging
+import warnings
 from collections import namedtuple
 from typing import TYPE_CHECKING, Callable, List, Optional, Tuple, Union
 
-from .endpoint import Endpoint, api
+from .endpoint import Endpoint, api, parameter_added_in
 from .exceptions import MissingRequiredFieldError
 from .. import RequestFactory, PaginationItem, ScheduleItem, TaskItem
 
 logger = logging.getLogger("tableau.endpoint.schedules")
-# Oh to have a first class Result concept in Python...
 AddResponse = namedtuple("AddResponse", ("result", "error", "warnings", "task_created"))
 OK = AddResponse(result=True, error=None, warnings=None, task_created=None)
 
 if TYPE_CHECKING:
     from ..request_options import RequestOptions
-    from ...models import DatasourceItem, WorkbookItem
+    from ...models import DatasourceItem, WorkbookItem, FlowItem
 
 
 class Schedules(Endpoint):
@@ -81,54 +81,71 @@ def create(self, schedule_item: ScheduleItem) -> ScheduleItem:
         return new_schedule
 
     @api(version="2.8")
+    @parameter_added_in(flow="3.3")
     def add_to_schedule(
         self,
         schedule_id: str,
         workbook: "WorkbookItem" = None,
         datasource: "DatasourceItem" = None,
-        task_type: str = TaskItem.Type.ExtractRefresh,
+        flow: "FlowItem" = None,
+        task_type: str = None,
     ) -> List[AddResponse]:
-        def add_to(
-            resource: Union["DatasourceItem", "WorkbookItem"],
-            type_: str,
-            req_factory: Callable[
-                [
-                    str,
-                    str,
-                ],
-                bytes,
-            ],
-        ) -> AddResponse:
-            id_ = resource.id
-            url = "{0}/{1}/{2}s".format(self.siteurl, schedule_id, type_)
-            add_req = req_factory(id_, task_type=task_type)  # type: ignore[call-arg, arg-type]
-            response = self.put_request(url, add_req)
-
-            error, warnings, task_created = ScheduleItem.parse_add_to_schedule_response(
-                response, self.parent_srv.namespace
-            )
-            if task_created:
-                logger.info("Added {} to {} to schedule {}".format(type_, id_, schedule_id))
 
-            if error is not None or warnings is not None:
-                return AddResponse(
-                    result=False,
-                    error=error,
-                    warnings=warnings,
-                    task_created=task_created,
-                )
-            else:
-                return OK
-
-        items = []
+        # There doesn't seem to be a good reason to allow one item of each type?
+        if workbook and datasource:
+            warnings.warn("Passing in multiple items for add_to_schedule will be deprecated", PendingDeprecationWarning)
+        items: List[
+            Tuple[str, Union[WorkbookItem, FlowItem, DatasourceItem], str, Callable[[Optional[str], str], bytes], str]
+        ] = []

         if workbook is not None:
-            items.append((workbook, "workbook", RequestFactory.Schedule.add_workbook_req))
+            if not task_type:
+                task_type = TaskItem.Type.ExtractRefresh
+            items.append((schedule_id, workbook, "workbook", RequestFactory.Schedule.add_workbook_req, task_type))
         if datasource is not None:
+            if not task_type:
+                task_type = TaskItem.Type.ExtractRefresh
+            items.append((schedule_id, datasource, "datasource", RequestFactory.Schedule.add_datasource_req, task_type))
+        if flow is not None and not (workbook or datasource):  # Cannot pass a flow with any other type
+            if not task_type:
+                task_type = TaskItem.Type.RunFlow
             items.append(
-                (datasource, "datasource", RequestFactory.Schedule.add_datasource_req)  # type:ignore[arg-type]
-            )
+                (schedule_id, flow, "flow", RequestFactory.Schedule.add_flow_req, task_type)
+            )  # type:ignore[arg-type]
 
-        results = (add_to(*x) for x in items)
+        results = (self._add_to(*x) for x in items)
         # list() is needed for python 3.x compatibility
         return list(filter(lambda x: not x.result, results))  # type:ignore[arg-type]
+
+    def _add_to(
+        self,
+        schedule_id,
+        resource: Union["DatasourceItem", "WorkbookItem", "FlowItem"],
+        type_: str,
+        req_factory: Callable[
+            [
+                str,
+                str,
+            ],
+            bytes,
+        ],
+        item_task_type,
+    ) -> AddResponse:
+        id_ = resource.id
+        url = "{0}/{1}/{2}s".format(self.siteurl, schedule_id, type_)
+        add_req = req_factory(id_, task_type=item_task_type)  # type: ignore[call-arg, arg-type]
+        response = self.put_request(url, add_req)
+
+        error, warnings, task_created = ScheduleItem.parse_add_to_schedule_response(response, self.parent_srv.namespace)
+        if task_created:
+            logger.info("Added {} to {} to schedule {}".format(type_, id_, schedule_id))
+
+        if error is not None or warnings is not None:
+            return AddResponse(
+                result=False,
+                error=error,
+                warnings=warnings,
+                task_created=task_created,
+            )
+        else:
+            return OK
diff --git a/tableauserverclient/server/endpoint/workbooks_endpoint.py b/tableauserverclient/server/endpoint/workbooks_endpoint.py
@@ -518,4 +518,9 @@ def delete_revision(self, workbook_id: str, revision_number: str) -> None:
         url = "/".join([self.baseurl, workbook_id, "revisions", revision_number])
 
         self.delete_request(url)
-        logger.info("Deleted single workbook revsision (ID: {0}) (Revision: {1})".format(workbook_id, revision_number))
+        logger.info("Deleted single workbook revision (ID: {0}) (Revision: {1})".format(workbook_id, revision_number))
+
+    # a convenience method
+    @api(version="2.8")
+    def schedule_extract_refresh(self, schedule_id: int, item: WorkbookItem) -> None:  # actually should return a task
+        return self.parent_srv.schedules.add_to_schedule(schedule_id, workbook=item)
diff --git a/tableauserverclient/server/request_factory.py b/tableauserverclient/server/request_factory.py
@@ -557,6 +557,9 @@ def add_workbook_req(self, id_: Optional[str], task_type: str = TaskItem.Type.Ex
     def add_datasource_req(self, id_: Optional[str], task_type: str = TaskItem.Type.ExtractRefresh) -> bytes:
         return self._add_to_req(id_, "datasource", task_type)
 
+    def add_flow_req(self, id_: Optional[str], task_type: str = TaskItem.Type.RunFlow) -> bytes:
+        return self._add_to_req(id_, "flow", task_type)
+
 
 class SiteRequest(object):
     def update_req(self, site_item: "SiteItem"):

diff --git a/test/assets/flow_get_by_id.xml b/test/assets/flow_get_by_id.xml
@@ -0,0 +1,10 @@
+<?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.5.xsd">
+    <flow id="587daa37-b84d-4400-a9a2-aa90e0be7837" name="FlowOne" description="Descriptive" webpageUrl="http://tableauserver/#/flows/1" fileType="tflx" createdAt="2019-06-16T21:43:28Z" updatedAt="2019-06-16T21:43:28Z">
+        <project id="aa23f4ac-906f-11e9-86fb-3f0f71412e77" name="Default"/>
+        <owner id="7ebb3f20-0fd2-4f27-a2f6-c539470999e2"/>
+        <tags>
+            <tag label="i_love_tags"/>
+        </tags>
+    </flow>
+</tsResponse>
diff --git a/test/assets/schedule_add_flow.xml b/test/assets/schedule_add_flow.xml
@@ -0,0 +1,9 @@
+<?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.3.xsd">
+  <task>
+    <runFlow>
+      <schedule id="schedule-id" />
+      <flow id="flow-id" />
+    </runFlow>
+  </task>
+</tsResponse>
diff --git a/test/test_schedule.py b/test/test_schedule.py
@@ -20,9 +20,11 @@
 ADD_WORKBOOK_TO_SCHEDULE = os.path.join(TEST_ASSET_DIR, "schedule_add_workbook.xml")
 ADD_WORKBOOK_TO_SCHEDULE_WITH_WARNINGS = os.path.join(TEST_ASSET_DIR, "schedule_add_workbook_with_warnings.xml")
 ADD_DATASOURCE_TO_SCHEDULE = os.path.join(TEST_ASSET_DIR, "schedule_add_datasource.xml")
+ADD_FLOW_TO_SCHEDULE = os.path.join(TEST_ASSET_DIR, "schedule_add_flow.xml")
 
 WORKBOOK_GET_BY_ID_XML = os.path.join(TEST_ASSET_DIR, "workbook_get_by_id.xml")
 DATASOURCE_GET_BY_ID_XML = os.path.join(TEST_ASSET_DIR, "datasource_get_by_id.xml")
+FLOW_GET_BY_ID_XML = os.path.join(TEST_ASSET_DIR, "flow_get_by_id.xml")
 
 
 class ScheduleTests(unittest.TestCase):
@@ -314,3 +316,18 @@ def test_add_datasource(self) -> None:
             datasource = self.server.datasources.get_by_id("bar")
             result = self.server.schedules.add_to_schedule("foo", datasource=datasource)
         self.assertEqual(0, len(result), "Added properly")
+
+    def test_add_flow(self) -> None:
+        self.server.version = "3.3"
+        baseurl = "{}/sites/{}/schedules".format(self.server.baseurl, self.server.site_id)
+
+        with open(FLOW_GET_BY_ID_XML, "rb") as f:
+            flow_response = f.read().decode("utf-8")
+        with open(ADD_FLOW_TO_SCHEDULE, "rb") as f:
+            add_flow_response = f.read().decode("utf-8")
+        with requests_mock.mock() as m:
+            m.get(self.server.flows.baseurl + "/bar", text=flow_response)
+            m.put(baseurl + "/foo/flows", text=flow_response)
+            flow = self.server.flows.get_by_id("bar")
+            result = self.server.schedules.add_to_schedule("foo", flow=flow)
+        self.assertEqual(0, len(result), "Added properly")