Add type hints for schedules

tableauserverclient/models/interval_item.py

@@ -84,8 +84,9 @@ def _interval_type_pairs(self):
 
 
 class DailyInterval(object):
-    def __init__(self, start_time):
+    def __init__(self, start_time, *interval_values):
         self.start_time = start_time
+        self.interval = interval_values
 
     @property
     def _frequency(self):
@@ -101,6 +102,14 @@ def start_time(self):
     def start_time(self, value):
         self._start_time = value
 
+    @property
+    def interval(self):
+        return self._interval
+
+    @interval.setter
+    def interval(self, interval):
+        self._interval = interval
+
 
 class WeeklyInterval(object):
     def __init__(self, start_time, *interval_values):

tableauserverclient/models/schedule_item.py

@@ -1,5 +1,6 @@
 import xml.etree.ElementTree as ET
 from datetime import datetime
+from typing import Optional, Union
 
 from .interval_item import (
     IntervalItem,
@@ -15,6 +16,8 @@
 )
 from ..datetime_helpers import parse_datetime
 
+Interval = Union[HourlyInterval, DailyInterval, WeeklyInterval, MonthlyInterval]
+
 
 class ScheduleItem(object):
     class Type:
@@ -31,86 +34,86 @@ class State:
         Active = "Active"
         Suspended = "Suspended"
 
-    def __init__(self, name, priority, schedule_type, execution_order, interval_item):
-        self._created_at = None
-        self._end_schedule_at = None
-        self._id = None
-        self._next_run_at = None
-        self._state = None
-        self._updated_at = None
-        self.interval_item = interval_item
-        self.execution_order = execution_order
-        self.name = name
-        self.priority = priority
-        self.schedule_type = schedule_type
+    def __init__(self, name: str, priority: int, schedule_type: str, execution_order: str, interval_item: Interval):
+        self._created_at: Optional[datetime] = None
+        self._end_schedule_at: Optional[datetime] = None
+        self._id: Optional[str] = None
+        self._next_run_at: Optional[datetime] = None
+        self._state: Optional[str] = None
+        self._updated_at: Optional[datetime] = None
+        self.interval_item: Interval = interval_item
+        self.execution_order: str = execution_order
+        self.name: str = name
+        self.priority: int = priority
+        self.schedule_type: str = schedule_type
 
     def __repr__(self):
-        return '<Schedule#{_id} "{_name}" {interval_item}>'.format(**self.__dict__)
+        return '<Schedule#{_id} "{_name}" {interval_item}>'.format(**vars(self))
 
     @property
-    def created_at(self):
+    def created_at(self) -> Optional[datetime]:
         return self._created_at
 
     @property
-    def end_schedule_at(self):
+    def end_schedule_at(self) -> Optional[datetime]:
         return self._end_schedule_at
 
     @property
-    def execution_order(self):
+    def execution_order(self) -> str:
         return self._execution_order
 
     @execution_order.setter
     @property_is_enum(ExecutionOrder)
-    def execution_order(self, value):
+    def execution_order(self, value: str):
         self._execution_order = value
 
     @property
-    def id(self):
+    def id(self) -> Optional[str]:
         return self._id
 
     @property
-    def name(self):
+    def name(self) -> str:
         return self._name
 
     @name.setter
     @property_not_nullable
-    def name(self, value):
+    def name(self, value: str):
         self._name = value
 
     @property
-    def next_run_at(self):
+    def next_run_at(self) -> Optional[datetime]:
         return self._next_run_at
 
     @property
-    def priority(self):
+    def priority(self) -> int:
         return self._priority
 
     @priority.setter
     @property_is_int(range=(1, 100))
-    def priority(self, value):
+    def priority(self, value: int):
         self._priority = value
 
     @property
-    def schedule_type(self):
+    def schedule_type(self) -> str:
         return self._schedule_type
 
     @schedule_type.setter
     @property_is_enum(Type)
     @property_not_nullable
-    def schedule_type(self, value):
+    def schedule_type(self, value: str):
         self._schedule_type = value
 
     @property
-    def state(self):
+    def state(self) -> Optional[str]:
         return self._state
 
     @state.setter
     @property_is_enum(State)
-    def state(self, value):
+    def state(self, value: str):
         self._state = value
 
     @property
-    def updated_at(self):
+    def updated_at(self) -> Optional[datetime]:
         return self._updated_at
 
     @property

tableauserverclient/server/endpoint/schedules_endpoint.py

@@ -4,24 +4,29 @@
 import logging
 import copy
 from collections import namedtuple
+from typing import TYPE_CHECKING, Callable, List, Optional, Tuple, Union
 
 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
+
 
 class Schedules(Endpoint):
     @property
-    def baseurl(self):
+    def baseurl(self) -> str:
         return "{0}/schedules".format(self.parent_srv.baseurl)
 
     @property
-    def siteurl(self):
+    def siteurl(self) -> str:
         return "{0}/sites/{1}/schedules".format(self.parent_srv.baseurl, self.parent_srv.site_id)
 
     @api(version="2.3")
-    def get(self, req_options=None):
+    def get(self, req_options: Optional["RequestOptions"] = None) -> Tuple[List[ScheduleItem], PaginationItem]:
         logger.info("Querying all schedules")
         url = self.baseurl
         server_response = self.get_request(url, req_options)
@@ -30,7 +35,7 @@ def get(self, req_options=None):
         return all_schedule_items, pagination_item
 
     @api(version="2.3")
-    def delete(self, schedule_id):
+    def delete(self, schedule_id: str) -> None:
         if not schedule_id:
             error = "Schedule ID undefined"
             raise ValueError(error)
@@ -39,7 +44,7 @@ def delete(self, schedule_id):
         logger.info("Deleted single schedule (ID: {0})".format(schedule_id))
 
     @api(version="2.3")
-    def update(self, schedule_item):
+    def update(self, schedule_item: ScheduleItem) -> ScheduleItem:
         if not schedule_item.id:
             error = "Schedule item missing ID."
             raise MissingRequiredFieldError(error)
@@ -52,7 +57,7 @@ def update(self, schedule_item):
         return updated_schedule._parse_common_tags(server_response.content, self.parent_srv.namespace)
 
     @api(version="2.3")
-    def create(self, schedule_item):
+    def create(self, schedule_item: ScheduleItem) -> ScheduleItem:
         if schedule_item.interval_item is None:
             error = "Interval item must be defined."
             raise MissingRequiredFieldError(error)
@@ -67,15 +72,25 @@ def create(self, schedule_item):
     @api(version="2.8")
     def add_to_schedule(
         self,
-        schedule_id,
-        workbook=None,
-        datasource=None,
-        task_type=TaskItem.Type.ExtractRefresh,
-    ):
-        def add_to(resource, type_, req_factory):
+        schedule_id: str,
+        workbook: "WorkbookItem" = None,
+        datasource: "DatasourceItem" = None,
+        task_type: str = TaskItem.Type.ExtractRefresh,
+    ) -> 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)
+            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(
@@ -99,8 +114,10 @@ def add_to(resource, type_, req_factory):
         if workbook is not None:
             items.append((workbook, "workbook", RequestFactory.Schedule.add_workbook_req))
         if datasource is not None:
-            items.append((datasource, "datasource", RequestFactory.Schedule.add_datasource_req))
+            items.append(
+                (datasource, "datasource", RequestFactory.Schedule.add_datasource_req)  # type:ignore[arg-type]
+            )
 
         results = (add_to(*x) for x in items)
         # list() is needed for python 3.x compatibility
-        return list(filter(lambda x: not x.result, results))
+        return list(filter(lambda x: not x.result, results))  # type:ignore[arg-type]

tableauserverclient/server/request_factory.py

@@ -515,7 +515,7 @@ def update_req(self, schedule_item):
                     single_interval_element.attrib[expression] = value
         return ET.tostring(xml_request)
 
-    def _add_to_req(self, id_, target_type, task_type=TaskItem.Type.ExtractRefresh):
+    def _add_to_req(self, id_: Optional[str], target_type: str, task_type: str = TaskItem.Type.ExtractRefresh) -> bytes:
         """
         <task>
           <target_type>
@@ -524,6 +524,8 @@ def _add_to_req(self, id_, target_type, task_type=TaskItem.Type.ExtractRefresh):
         </task>
 
         """
+        if not isinstance(id_, str):
+            raise ValueError(f"id_ should be a string, reeceived: {type(id_)}")
         xml_request = ET.Element("tsRequest")
         task_element = ET.SubElement(xml_request, "task")
         task = ET.SubElement(task_element, task_type)
@@ -532,10 +534,10 @@ def _add_to_req(self, id_, target_type, task_type=TaskItem.Type.ExtractRefresh):
 
         return ET.tostring(xml_request)
 
-    def add_workbook_req(self, id_, task_type=TaskItem.Type.ExtractRefresh):
+    def add_workbook_req(self, id_: Optional[str], task_type: str = TaskItem.Type.ExtractRefresh) -> bytes:
         return self._add_to_req(id_, "workbook", task_type)
 
-    def add_datasource_req(self, id_, task_type=TaskItem.Type.ExtractRefresh):
+    def add_datasource_req(self, id_: Optional[str], task_type: str = TaskItem.Type.ExtractRefresh) -> bytes:
         return self._add_to_req(id_, "datasource", task_type)