feat: parse linked task xml

Author: jorwoods

tableauserverclient/models/linked_tasks_item.py

@@ -0,0 +1,74 @@
+from typing import List, Optional
+from defusedxml.ElementTree import fromstring
+
+from tableauserverclient.models.schedule_item import ScheduleItem
+from tableauserverclient.models.task_item import TaskItem
+
+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
+
+
+
+def string_to_bool(s: str) -> bool:
+    return s.lower() == "true"

tableauserverclient/server/endpoint/linked_tasks_endpoint.py

@@ -0,0 +1,17 @@
+from typing import Optional
+from tableauserverclient.server.endpoint.endpoint import QuerysetEndpoint, api
+from tableauserverclient.server.request_options import RequestOptions
+
+class LinkedTasks(QuerysetEndpoint):
+    def __init__(self, parent_srv):
+        super().__init__(parent_srv)
+        self._parent_srv = parent_srv
+
+    @property
+    def baseurl(self):
+        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):
+        ...
+

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/test_linked_tasks.py

@@ -0,0 +1,62 @@
+from pathlib import Path
+import unittest
+
+from defusedxml.ElementTree import fromstring
+import pytest
+
+import tableauserverclient as TSC
+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"
+
+class TestLinkedTasks(unittest.TestCase):
+    
+    def setUp(self) -> None:
+        self.server = TSC.Server("http://test", False)
+
+        # 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)
+        self.assertEqual(1, len(task_runs))
+        task = task_runs[0]
+        self.assertEqual(task.flow_run_id, "e3d1fc25-5644-4e32-af35-58dcbd1dbd73")
+        self.assertEqual(task.flow_run_priority, 1)
+        self.assertEqual(task.flow_run_consecutive_failed_count, 3)
+        self.assertEqual(task.flow_run_task_type, "runFlow")
+        self.assertEqual(task.flow_id, "ab1231eb-b8ca-461e-a131-83f3c2b6a673")
+        self.assertEqual(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)
+        self.assertEqual(1, len(steps))
+        step = steps[0]
+        self.assertEqual(step.id, "f554a4df-bb6f-4294-94ee-9a709ef9bda0")
+        self.assertTrue(step.stop_downstream_on_failure)
+        self.assertEqual(step.step_number, 1)
+        self.assertEqual(1, len(step.task_details))
+        task = step.task_details[0]
+        self.assertEqual(task.flow_run_id, "e3d1fc25-5644-4e32-af35-58dcbd1dbd73")
+        self.assertEqual(task.flow_run_priority, 1)
+        self.assertEqual(task.flow_run_consecutive_failed_count, 3)
+        self.assertEqual(task.flow_run_task_type, "runFlow")
+        self.assertEqual(task.flow_id, "ab1231eb-b8ca-461e-a131-83f3c2b6a673")
+        self.assertEqual(task.flow_name, "flow-name")
+
+    def test_parse_linked_task(self):
+        tasks = LinkedTaskItem.from_response(GET_LINKED_TASKS.read_bytes(), self.server.namespace)
+        self.assertEqual(1, len(tasks))
+        task = tasks[0]
+        self.assertEqual(task.id, "1b8211dc-51a8-45ce-a831-b5921708e03e")
+        self.assertEqual(task.num_steps, 1)
+        self.assertEqual(task.schedule.id, "be077332-d01d-481b-b2f3-917e463d4dca")
+