feat: linked tasks

tableauserverclient/__init__.py

@@ -22,6 +22,9 @@
     IntervalItem,
     JobItem,
     JWTAuth,
+    LinkedTaskItem,
+    LinkedTaskStepItem,
+    LinkedTaskFlowRunItem,
     MetricItem,
     MonthlyInterval,
     PaginationItem,
@@ -118,4 +121,7 @@
     "Pager",
     "Server",
     "Sort",
+    "LinkedTaskItem",
+    "LinkedTaskStepItem",
+    "LinkedTaskFlowRunItem",
 ]

tableauserverclient/models/__init__.py

@@ -23,6 +23,11 @@
     HourlyInterval,
 )
 from tableauserverclient.models.job_item import JobItem, BackgroundJobItem
+from tableauserverclient.models.linked_tasks_item import (
+    LinkedTaskItem,
+    LinkedTaskStepItem,
+    LinkedTaskFlowRunItem,
+)
 from tableauserverclient.models.metric_item import MetricItem
 from tableauserverclient.models.pagination_item import PaginationItem
 from tableauserverclient.models.permissions_item import PermissionsRule, Permission
@@ -93,4 +98,7 @@
     "ViewItem",
     "WebhookItem",
     "WorkbookItem",
+    "LinkedTaskItem",
+    "LinkedTaskStepItem",
+    "LinkedTaskFlowRunItem",
 ]

tableauserverclient/models/linked_tasks_item.py

@@ -0,0 +1,102 @@
+import datetime as dt
+from typing import List, Optional
+
+from defusedxml.ElementTree import fromstring
+
+from tableauserverclient.datetime_helpers import parse_datetime
+from tableauserverclient.models.schedule_item import ScheduleItem
+
+
+class LinkedTaskItem:
+    def __init__(self) -> None:
+        self.id: Optional[str] = None
+        self.num_steps: Optional[int] = None
+        self.schedule: Optional[ScheduleItem] = None
+
+    @classmethod
+    def from_response(cls, resp: bytes, namespace) -> List["LinkedTaskItem"]:
+        parsed_response = fromstring(resp)
+        return [
+            cls._parse_element(x, namespace)
+            for x in parsed_response.findall(".//t:linkedTasks[@id]", namespaces=namespace)
+        ]
+
+    @classmethod
+    def _parse_element(cls, xml, namespace) -> "LinkedTaskItem":
+        task = cls()
+        task.id = xml.get("id")
+        task.num_steps = int(xml.get("numSteps"))
+        task.schedule = ScheduleItem.from_element(xml, namespace)[0]
+        return task
+
+
+class LinkedTaskStepItem:
+    def __init__(self) -> None:
+        self.id: Optional[str] = None
+        self.step_number: Optional[int] = None
+        self.stop_downstream_on_failure: Optional[bool] = None
+        self.task_details: List[LinkedTaskFlowRunItem] = []
+
+    @classmethod
+    def from_task_xml(cls, xml, namespace) -> List["LinkedTaskStepItem"]:
+        return [cls._parse_element(x, namespace) for x in xml.findall(".//t:linkedTaskSteps[@id]", namespace)]
+
+    @classmethod
+    def _parse_element(cls, xml, namespace) -> "LinkedTaskStepItem":
+        step = cls()
+        step.id = xml.get("id")
+        step.step_number = int(xml.get("stepNumber"))
+        step.stop_downstream_on_failure = string_to_bool(xml.get("stopDownstreamTasksOnFailure"))
+        step.task_details = LinkedTaskFlowRunItem._parse_element(xml, namespace)
+        return step
+
+
+class LinkedTaskFlowRunItem:
+    def __init__(self) -> None:
+        self.flow_run_id: Optional[str] = None
+        self.flow_run_priority: Optional[int] = None
+        self.flow_run_consecutive_failed_count: Optional[int] = None
+        self.flow_run_task_type: Optional[str] = None
+        self.flow_id: Optional[str] = None
+        self.flow_name: Optional[str] = None
+
+    @classmethod
+    def _parse_element(cls, xml, namespace) -> List["LinkedTaskFlowRunItem"]:
+        all_tasks = []
+        for flow_run in xml.findall(".//t:flowRun[@id]", namespace):
+            task = cls()
+            task.flow_run_id = flow_run.get("id")
+            task.flow_run_priority = int(flow_run.get("priority"))
+            task.flow_run_consecutive_failed_count = int(flow_run.get("consecutiveFailedCount"))
+            task.flow_run_task_type = flow_run.get("type")
+            flow = flow_run.find(".//t:flow[@id]", namespace)
+            task.flow_id = flow.get("id")
+            task.flow_name = flow.get("name")
+            all_tasks.append(task)
+
+        return all_tasks
+
+
+class LinkedTaskJobItem:
+    def __init__(self) -> None:
+        self.id: Optional[str] = None
+        self.linked_task_id: Optional[str] = None
+        self.status: Optional[str] = None
+        self.created_at: Optional[dt.datetime] = None
+
+    @classmethod
+    def from_response(cls, resp: bytes, namespace) -> "LinkedTaskJobItem":
+        parsed_response = fromstring(resp)
+        job = cls()
+        job_xml = parsed_response.find(".//t:linkedTaskJob[@id]", namespaces=namespace)
+        if job_xml is None:
+            raise ValueError("No linked task job found in response")
+        job.id = job_xml.get("id")
+        job.linked_task_id = job_xml.get("linkedTaskId")
+        job.status = job_xml.get("status")
+        job.created_at = parse_datetime(job_xml.get("createdAt"))
+        return job
+
+
+def string_to_bool(s: str) -> bool:
+    return s.lower() == "true"

tableauserverclient/server/endpoint/__init__.py

@@ -14,6 +14,7 @@
 from tableauserverclient.server.endpoint.groups_endpoint import Groups
 from tableauserverclient.server.endpoint.groupsets_endpoint import GroupSets
 from tableauserverclient.server.endpoint.jobs_endpoint import Jobs
+from tableauserverclient.server.endpoint.linked_tasks_endpoint import LinkedTasks
 from tableauserverclient.server.endpoint.metadata_endpoint import Metadata
 from tableauserverclient.server.endpoint.metrics_endpoint import Metrics
 from tableauserverclient.server.endpoint.projects_endpoint import Projects
@@ -46,6 +47,7 @@
     "Groups",
     "GroupSets",
     "Jobs",
+    "LinkedTasks",
     "Metadata",
     "Metrics",
     "Projects",

tableauserverclient/server/endpoint/linked_tasks_endpoint.py

@@ -0,0 +1,45 @@
+from typing import List, Optional, Tuple, Union
+
+from tableauserverclient.helpers.logging import logger
+from tableauserverclient.models.linked_tasks_item import LinkedTaskItem, LinkedTaskJobItem
+from tableauserverclient.models.pagination_item import PaginationItem
+from tableauserverclient.server.endpoint.endpoint import QuerysetEndpoint, api
+from tableauserverclient.server.request_factory import RequestFactory
+from tableauserverclient.server.request_options import RequestOptions
+
+
+class LinkedTasks(QuerysetEndpoint[LinkedTaskItem]):
+    def __init__(self, parent_srv):
+        super().__init__(parent_srv)
+        self._parent_srv = parent_srv
+
+    @property
+    def baseurl(self) -> str:
+        return f"{self.parent_srv.baseurl}/sites/{self.parent_srv.site_id}/tasks/linked"
+
+    @api(version="3.15")
+    def get(self, req_options: Optional["RequestOptions"] = None) -> Tuple[List[LinkedTaskItem], PaginationItem]:
+        logger.info("Querying all linked tasks on site")
+        url = self.baseurl
+        server_response = self.get_request(url, req_options)
+        pagination_item = PaginationItem.from_response(server_response.content, self.parent_srv.namespace)
+        all_group_items = LinkedTaskItem.from_response(server_response.content, self.parent_srv.namespace)
+        return all_group_items, pagination_item
+
+    @api(version="3.15")
+    def get_by_id(self, linked_task: Union[LinkedTaskItem, str]) -> LinkedTaskItem:
+        task_id = getattr(linked_task, "id", linked_task)
+        logger.info("Querying all linked tasks on site")
+        url = f"{self.baseurl}/{task_id}"
+        server_response = self.get_request(url)
+        all_group_items = LinkedTaskItem.from_response(server_response.content, self.parent_srv.namespace)
+        return all_group_items[0]
+
+    @api(version="3.15")
+    def run_now(self, linked_task: Union[LinkedTaskItem, str]) -> LinkedTaskJobItem:
+        task_id = getattr(linked_task, "id", linked_task)
+        logger.info(f"Running linked task {task_id} now")
+        url = f"{self.baseurl}/{task_id}/runNow"
+        empty_req = RequestFactory.Empty.empty_req()
+        server_response = self.post_request(url, empty_req)
+        return LinkedTaskJobItem.from_response(server_response.content, self.parent_srv.namespace)

tableauserverclient/server/server.py

@@ -33,6 +33,7 @@
     Metrics,
     Endpoint,
     CustomViews,
+    LinkedTasks,
     GroupSets,
 )
 from tableauserverclient.server.exceptions import (
@@ -100,6 +101,7 @@ def __init__(self, server_address, use_server_version=False, http_options=None,
         self.flow_runs = FlowRuns(self)
         self.metrics = Metrics(self)
         self.custom_views = CustomViews(self)
+        self.linked_tasks = LinkedTasks(self)
         self.group_sets = GroupSets(self)
 
         self._session = self._session_factory()

test/assets/linked_tasks_get.xml

@@ -0,0 +1,33 @@
+<?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">
+   <pagination pageNumber="1" pageSize="100" totalAvailable="5"/>
+   <linkedTasks>
+     <linkedTasks id="1b8211dc-51a8-45ce-a831-b5921708e03e"
+      numSteps="1">
+       <schedule id="be077332-d01d-481b-b2f3-917e463d4dca"
+        name="schedule-name"
+        state="Active"
+        priority="50"
+        createdAt="2024-07-24T00:27:55Z"
+        updatedAt="2024-07-24T01:42:15Z"
+        type="Flow"
+        frequency="daily"
+        nextRunAt="2024-07-24T03:30:00Z"/>
+       <linkedTaskSteps>
+         <linkedTaskSteps id="f554a4df-bb6f-4294-94ee-9a709ef9bda0"
+          stepNumber="1"
+          stopDownstreamTasksOnFailure="true">
+         <task>
+           <flowRun id="e3d1fc25-5644-4e32-af35-58dcbd1dbd73"
+            priority="1"
+            consecutiveFailedCount="3"
+            type="runFlow">
+                <flow id="ab1231eb-b8ca-461e-a131-83f3c2b6a673"
+                name="flow-name" />
+           </flowRun>
+         </task>
+       </linkedTaskSteps>
+     </linkedTaskSteps>
+    </linkedTasks>
+    </linkedTasks>
+</tsResponse>

test/assets/linked_tasks_run_now.xml

@@ -0,0 +1,7 @@
+<?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">
+<linkedTaskJob id="269a1e5a-1220-4a13-ac01-704982693dd8"
+    linkedTaskId="1b8211dc-51a8-45ce-a831-b5921708e03e"
+    status="InProgress"
+    createdAt="2022-02-15T00:22:22Z"/>
+  </tsResponse>

test/test_linked_tasks.py

@@ -0,0 +1,129 @@
+from pathlib import Path
+import unittest
+
+from defusedxml.ElementTree import fromstring
+import pytest
+import requests_mock
+
+import tableauserverclient as TSC
+from tableauserverclient.datetime_helpers import parse_datetime
+from tableauserverclient.models.linked_tasks_item import LinkedTaskItem, LinkedTaskStepItem, LinkedTaskFlowRunItem
+
+asset_dir = (Path(__file__).parent / "assets").resolve()
+
+GET_LINKED_TASKS = asset_dir / "linked_tasks_get.xml"
+RUN_LINKED_TASK_NOW = asset_dir / "linked_tasks_run_now.xml"
+
+
+class TestLinkedTasks(unittest.TestCase):
+    def setUp(self) -> None:
+        self.server = TSC.Server("http://test", False)
+        self.server.version = "3.15"
+
+        # Fake signin
+        self.server._site_id = "dad65087-b08b-4603-af4e-2887b8aafc67"
+        self.server._auth_token = "j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM"
+
+        self.baseurl = self.server.linked_tasks.baseurl
+
+    def test_parse_linked_task_flow_run(self):
+        xml = fromstring(GET_LINKED_TASKS.read_bytes())
+        task_runs = LinkedTaskFlowRunItem._parse_element(xml, self.server.namespace)
+        assert 1 == len(task_runs)
+        task = task_runs[0]
+        assert task.flow_run_id == "e3d1fc25-5644-4e32-af35-58dcbd1dbd73"
+        assert task.flow_run_priority == 1
+        assert task.flow_run_consecutive_failed_count == 3
+        assert task.flow_run_task_type == "runFlow"
+        assert task.flow_id == "ab1231eb-b8ca-461e-a131-83f3c2b6a673"
+        assert task.flow_name == "flow-name"
+
+    def test_parse_linked_task_step(self):
+        xml = fromstring(GET_LINKED_TASKS.read_bytes())
+        steps = LinkedTaskStepItem.from_task_xml(xml, self.server.namespace)
+        assert 1 == len(steps)
+        step = steps[0]
+        assert step.id == "f554a4df-bb6f-4294-94ee-9a709ef9bda0"
+        assert step.stop_downstream_on_failure
+        assert step.step_number == 1
+        assert 1 == len(step.task_details)
+        task = step.task_details[0]
+        assert task.flow_run_id == "e3d1fc25-5644-4e32-af35-58dcbd1dbd73"
+        assert task.flow_run_priority == 1
+        assert task.flow_run_consecutive_failed_count == 3
+        assert task.flow_run_task_type == "runFlow"
+        assert task.flow_id == "ab1231eb-b8ca-461e-a131-83f3c2b6a673"
+        assert task.flow_name == "flow-name"
+
+    def test_parse_linked_task(self):
+        tasks = LinkedTaskItem.from_response(GET_LINKED_TASKS.read_bytes(), self.server.namespace)
+        assert 1 == len(tasks)
+        task = tasks[0]
+        assert task.id == "1b8211dc-51a8-45ce-a831-b5921708e03e"
+        assert task.num_steps == 1
+        assert task.schedule is not None
+        assert task.schedule.id == "be077332-d01d-481b-b2f3-917e463d4dca"
+
+    def test_get_linked_tasks(self):
+        with requests_mock.mock() as m:
+            m.get(self.baseurl, text=GET_LINKED_TASKS.read_text())
+            tasks, pagination_item = self.server.linked_tasks.get()
+
+        assert 1 == len(tasks)
+        task = tasks[0]
+        assert task.id == "1b8211dc-51a8-45ce-a831-b5921708e03e"
+        assert task.num_steps == 1
+        assert task.schedule is not None
+        assert task.schedule.id == "be077332-d01d-481b-b2f3-917e463d4dca"
+
+    def test_get_by_id_str_linked_task(self):
+        id_ = "1b8211dc-51a8-45ce-a831-b5921708e03e"
+
+        with requests_mock.mock() as m:
+            m.get(f"{self.baseurl}/{id_}", text=GET_LINKED_TASKS.read_text())
+            task = self.server.linked_tasks.get_by_id(id_)
+
+        assert task.id == "1b8211dc-51a8-45ce-a831-b5921708e03e"
+        assert task.num_steps == 1
+        assert task.schedule is not None
+        assert task.schedule.id == "be077332-d01d-481b-b2f3-917e463d4dca"
+
+    def test_get_by_id_obj_linked_task(self):
+        id_ = "1b8211dc-51a8-45ce-a831-b5921708e03e"
+        in_task = LinkedTaskItem()
+        in_task.id = id_
+
+        with requests_mock.mock() as m:
+            m.get(f"{self.baseurl}/{id_}", text=GET_LINKED_TASKS.read_text())
+            task = self.server.linked_tasks.get_by_id(in_task)
+
+        assert task.id == "1b8211dc-51a8-45ce-a831-b5921708e03e"
+        assert task.num_steps == 1
+        assert task.schedule is not None
+        assert task.schedule.id == "be077332-d01d-481b-b2f3-917e463d4dca"
+
+    def test_run_now_str_linked_task(self):
+        id_ = "1b8211dc-51a8-45ce-a831-b5921708e03e"
+
+        with requests_mock.mock() as m:
+            m.post(f"{self.baseurl}/{id_}/runNow", text=RUN_LINKED_TASK_NOW.read_text())
+            job = self.server.linked_tasks.run_now(id_)
+
+        assert job.id == "269a1e5a-1220-4a13-ac01-704982693dd8"
+        assert job.status == "InProgress"
+        assert job.created_at == parse_datetime("2022-02-15T00:22:22Z")
+        assert job.linked_task_id == id_
+
+    def test_run_now_obj_linked_task(self):
+        id_ = "1b8211dc-51a8-45ce-a831-b5921708e03e"
+        in_task = LinkedTaskItem()
+        in_task.id = id_
+
+        with requests_mock.mock() as m:
+            m.post(f"{self.baseurl}/{id_}/runNow", text=RUN_LINKED_TASK_NOW.read_text())
+            job = self.server.linked_tasks.run_now(in_task)
+
+        assert job.id == "269a1e5a-1220-4a13-ac01-704982693dd8"
+        assert job.status == "InProgress"
+        assert job.created_at == parse_datetime("2022-02-15T00:22:22Z")
+        assert job.linked_task_id == id_