determine what datasources or workbooks are associated with a schedule

Author: casey-crawford-cfa

tableauserverclient/models/__init__.py

@@ -48,6 +48,7 @@
 from tableauserverclient.models.virtual_connection_item import VirtualConnectionItem
 from tableauserverclient.models.webhook_item import WebhookItem
 from tableauserverclient.models.workbook_item import WorkbookItem
+from tableauserverclient.models.extract_item import ExtractItem
 
 __all__ = [
     "ColumnItem",
@@ -103,4 +104,5 @@
     "LinkedTaskItem",
     "LinkedTaskStepItem",
     "LinkedTaskFlowRunItem",
+    "ExtractItem",
 ]

tableauserverclient/models/extract_item.py

@@ -0,0 +1,86 @@
+from typing import Optional, List
+from defusedxml.ElementTree import fromstring
+import xml.etree.ElementTree as ET
+
+
+class ExtractItem:
+    """
+    An extract refresh task item.
+    
+    Attributes
+    ----------
+    id : str
+        The ID of the extract refresh task
+    priority : int
+        The priority of the task
+    type : str
+        The type of extract refresh (incremental or full)
+    workbook_id : str, optional
+        The ID of the workbook if this is a workbook extract
+    datasource_id : str, optional
+        The ID of the datasource if this is a datasource extract
+    """
+
+    def __init__(
+        self,
+        priority: int,
+        refresh_type: str,
+        workbook_id: Optional[str] = None,
+        datasource_id: Optional[str] = None
+    ):
+        self._id: Optional[str] = None
+        self._priority = priority
+        self._type = refresh_type
+        self._workbook_id = workbook_id
+        self._datasource_id = datasource_id
+
+    @property
+    def id(self) -> Optional[str]:
+        return self._id
+
+    @property
+    def priority(self) -> int:
+        return self._priority
+
+    @property
+    def type(self) -> str:
+        return self._type
+
+    @property
+    def workbook_id(self) -> Optional[str]:
+        return self._workbook_id
+
+    @property
+    def datasource_id(self) -> Optional[str]:
+        return self._datasource_id
+
+    @classmethod
+    def from_response(cls, resp: str, ns: dict) -> List["ExtractItem"]:
+        """Create ExtractItem objects from XML response."""
+        parsed_response = fromstring(resp)
+        return cls.from_xml_element(parsed_response, ns)
+
+    @classmethod
+    def from_xml_element(cls, parsed_response: ET.Element, ns: dict) -> List["ExtractItem"]:
+        """Create ExtractItem objects from XML element."""
+        all_extract_items = []
+        all_extract_xml = parsed_response.findall(".//t:extract", namespaces=ns)
+        
+        for extract_xml in all_extract_xml:
+            extract_id = extract_xml.get("id", None)
+            priority = int(extract_xml.get("priority", 0))
+            refresh_type = extract_xml.get("type", None)
+            
+            # Check for workbook or datasource
+            workbook_elem = extract_xml.find(".//t:workbook", namespaces=ns)
+            datasource_elem = extract_xml.find(".//t:datasource", namespaces=ns)
+            
+            workbook_id = workbook_elem.get("id", None) if workbook_elem is not None else None
+            datasource_id = datasource_elem.get("id", None) if datasource_elem is not None else None
+            
+            extract_item = cls(priority, refresh_type, workbook_id, datasource_id)
+            extract_item._id = extract_id
+            
+            all_extract_items.append(extract_item)
+            
+        return all_extract_items

tableauserverclient/server/endpoint/schedules_endpoint.py

@@ -7,7 +7,7 @@
 from .endpoint import Endpoint, api, parameter_added_in
 from .exceptions import MissingRequiredFieldError
 from tableauserverclient.server import RequestFactory
-from tableauserverclient.models import PaginationItem, ScheduleItem, TaskItem
+from tableauserverclient.models import PaginationItem, ScheduleItem, TaskItem, ExtractItem
 
 from tableauserverclient.helpers.logging import logger
 
@@ -149,3 +149,19 @@ def _add_to(
             )
         else:
             return OK
+
+    @api(version="2.3")
+    def get_extract_refresh_tasks(self, schedule_id: str, req_options: Optional["RequestOptions"] = None) -> tuple[list["ExtractItem"], "PaginationItem"]:
+        """Get all extract refresh tasks for the specified schedule."""
+        if not schedule_id:
+            error = "Schedule ID undefined"
+            raise ValueError(error)
+        
+        logger.info(f"Querying extract refresh tasks for schedule (ID: {schedule_id})")
+        url = f"{self.siteurl}/{schedule_id}/extracts"
+        server_response = self.get_request(url, req_options)
+        
+        pagination_item = PaginationItem.from_response(server_response.content, self.parent_srv.namespace)
+        extract_items = ExtractItem.from_response(server_response.content, self.parent_srv.namespace)
+        
+        return extract_items, pagination_item

test/assets/schedule_get_extract_refresh_tasks.xml

@@ -0,0 +1,15 @@
+<?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">
+  <extracts>
+    <extract id="task1"
+      priority="1"
+      type="incremental-or-full" >
+        <workbook id="workbook-id" />
+    </extract>
+    <extract id="task2"
+      priority="2"
+      type="incremental-or-full" >
+        <datasource id="datasource-id" />
+    </extract>
+  </extracts>
+</tsResponse>

test/test_schedule.py

@@ -25,6 +25,7 @@
 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")
+GET_EXTRACT_TASKS_XML = os.path.join(TEST_ASSET_DIR, "schedule_get_extract_refresh_tasks.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")
@@ -405,3 +406,24 @@ def test_add_flow(self) -> None:
             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")
+
+    def test_get_extract_refresh_tasks(self) -> None:
+        self.server.version = "2.3"
+        
+
+        with open(GET_EXTRACT_TASKS_XML, "rb") as f:
+            response_xml = f.read().decode("utf-8")
+        with requests_mock.mock() as m:
+            schedule_id = "c9cff7f9-309c-4361-99ff-d4ba8c9f5467"
+            # baseurl = f"{self.baseurl}/schedules/{schedule_id}/extracts"
+            baseurl = f"{self.server.baseurl}/sites/{self.server.site_id}/schedules/{schedule_id}/extracts"
+            # Fix the URL construction to match the endpoint pattern
+            # url = f"{self.baseurl}/{schedule_id}/extracts"
+            m.get(baseurl, text=response_xml)
+            
+            extracts = self.server.schedules.get_extract_refresh_tasks(schedule_id)
+
+            self.assertIsNotNone(extracts)
+            self.assertIsInstance(extracts[0], list)
+            self.assertEqual(2, len(extracts[0]))
+            self.assertEqual("task1", extracts[0][0].id)