feat: get linked tasks

Author: jorwoods

tableauserverclient/__init__.py

@@ -21,6 +21,9 @@
     IntervalItem,
     JobItem,
     JWTAuth,
+    LinkedTaskItem,
+    LinkedTaskStepItem,
+    LinkedTaskFlowRunItem,
     MetricItem,
     MonthlyInterval,
     PaginationItem,
@@ -116,4 +119,8 @@
     "Pager",
     "Server",
     "Sort",
+    "LinkedTaskItem",
+    "LinkedTaskStepItem",
+    "LinkedTaskFlowRunItem",
+
 ]

tableauserverclient/models/__init__.py

@@ -22,6 +22,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
@@ -91,4 +96,7 @@
     "ViewItem",
     "WebhookItem",
     "WorkbookItem",
+    "LinkedTaskItem",
+    "LinkedTaskStepItem",
+    "LinkedTaskFlowRunItem",
 ]

tableauserverclient/models/linked_tasks_item.py

@@ -2,7 +2,6 @@
 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:

tableauserverclient/server/endpoint/__init__.py

@@ -13,6 +13,7 @@
 from tableauserverclient.server.endpoint.flow_task_endpoint import FlowTasks
 from tableauserverclient.server.endpoint.groups_endpoint import Groups
 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
@@ -44,6 +45,7 @@
     "FlowTasks",
     "Groups",
     "Jobs",
+    "LinkedTasks",
     "Metadata",
     "Metrics",
     "Projects",

tableauserverclient/server/endpoint/linked_tasks_endpoint.py

@@ -1,17 +1,26 @@
-from typing import Optional
+from typing import List, Optional, Tuple
+
+from tableauserverclient.helpers.logging import logger
+from tableauserverclient.models.linked_tasks_item import LinkedTaskItem
+from tableauserverclient.models.pagination_item import PaginationItem
 from tableauserverclient.server.endpoint.endpoint import QuerysetEndpoint, api
 from tableauserverclient.server.request_options import RequestOptions
 
-class LinkedTasks(QuerysetEndpoint):
+class LinkedTasks(QuerysetEndpoint[LinkedTaskItem]):
     def __init__(self, parent_srv):
         super().__init__(parent_srv)
         self._parent_srv = parent_srv
 
     @property
-    def baseurl(self):
+    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):
-        ...
+    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
 

tableauserverclient/server/server.py

@@ -33,6 +33,7 @@
     Metrics,
     Endpoint,
     CustomViews,
+    LinkedTasks,
 )
 from tableauserverclient.server.exceptions import (
     ServerInfoEndpointNotFoundError,
@@ -99,6 +100,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._session = self._session_factory()
         self._http_options = dict()  # must set this before making a server call

test/test_linked_tasks.py

@@ -3,6 +3,7 @@
 
 from defusedxml.ElementTree import fromstring
 import pytest
+import requests_mock
 
 import tableauserverclient as TSC
 from tableauserverclient.models.linked_tasks_item import LinkedTaskItem, LinkedTaskStepItem, LinkedTaskFlowRunItem
@@ -15,12 +16,13 @@ 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
+        self.baseurl = self.server.linked_tasks.baseurl
 
     def test_parse_linked_task_flow_run(self):
         xml = fromstring(GET_LINKED_TASKS.read_bytes())
@@ -60,3 +62,15 @@ def test_parse_linked_task(self):
         self.assertEqual(task.num_steps, 1)
         self.assertEqual(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()
+
+        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")
+
+