Implement custom view objects (#1195)

* create custom views item and endpoint with get/update/delete methods
Also added custom view operations to the workbook sample

Author: jacalata

samples/explore_workbook.py

@@ -74,6 +74,10 @@ def main():
         if all_workbooks:
             # Pick one workbook from the list
             sample_workbook = all_workbooks[0]
+            sample_workbook.name = "Name me something cooler"
+            sample_workbook.description = "That doesn't work"
+            updated: TSC.WorkbookItem = server.workbooks.update(sample_workbook)
+            print(updated.name, updated.description)
 
             # Populate views
             server.workbooks.populate_views(sample_workbook)
@@ -127,6 +131,31 @@ def main():
                     f.write(sample_workbook.preview_image)
                 print("\nDownloaded preview image of workbook to {}".format(os.path.abspath(args.preview_image)))
 
+            # get custom views
+            cvs, _ = server.custom_views.get()
+            for c in cvs:
+                print(c)
+
+            # for the last custom view in the list
+
+            # update the name
+            # note that this will fail if the name is already changed to this value
+            changed: TSC.CustomViewItem(id=c.id, name="I was updated by tsc")
+            verified_change = server.custom_views.update(changed)
+            print(verified_change)
+
+            # export as image. Filters etc could be added here as usual
+            server.custom_views.populate_image(c)
+            filename = c.id + "-image-export.png"
+            with open(filename, "wb") as f:
+                f.write(c.image)
+            print("saved to " + filename)
+
+            if args.delete:
+                print("deleting {}".format(c.id))
+                unlucky = TSC.CustomViewItem(c.id)
+                server.custom_views.delete(unlucky.id)
+
 
 if __name__ == "__main__":
     main()

tableauserverclient/models/__init__.py

@@ -1,6 +1,7 @@
 from .column_item import ColumnItem
 from .connection_credentials import ConnectionCredentials
 from .connection_item import ConnectionItem
+from .custom_view_item import CustomViewItem
 from .data_acceleration_report_item import DataAccelerationReportItem
 from .data_alert_item import DataAlertItem
 from .database_item import DatabaseItem

tableauserverclient/models/custom_view_item.py

@@ -0,0 +1,156 @@
+from datetime import datetime
+
+from defusedxml import ElementTree
+from defusedxml.ElementTree import fromstring, tostring
+from typing import Callable, List, Optional
+
+from .exceptions import UnpopulatedPropertyError
+from .user_item import UserItem
+from .view_item import ViewItem
+from .workbook_item import WorkbookItem
+from ..datetime_helpers import parse_datetime
+
+
+class CustomViewItem(object):
+    def __init__(self, id: Optional[str] = None, name: Optional[str] = None) -> None:
+        self._content_url: Optional[str] = None  # ?
+        self._created_at: Optional["datetime"] = None
+        self._id: Optional[str] = id
+        self._image: Optional[Callable[[], bytes]] = None
+        self._name: Optional[str] = name
+        self._shared: Optional[bool] = False
+        self._updated_at: Optional["datetime"] = None
+
+        self._owner: Optional[UserItem] = None
+        self._view: Optional[ViewItem] = None
+        self._workbook: Optional[WorkbookItem] = None
+
+    def __repr__(self: "CustomViewItem"):
+        view_info = ""
+        if self._view:
+            view_info = " view='{}'".format(self._view.name or self._view.id or "unknown")
+        wb_info = ""
+        if self._workbook:
+            wb_info = " workbook='{}'".format(self._workbook.name or self._workbook.id or "unknown")
+        owner_info = ""
+        if self._owner:
+            owner_info = " owner='{}'".format(self._owner.name or self._owner.id or "unknown")
+        return "<CustomViewItem id={} name=`{}`{}{}{}>".format(self.id, self.name, view_info, wb_info, owner_info)
+
+    def _set_image(self, image):
+        self._image = image
+
+    @property
+    def content_url(self) -> Optional[str]:
+        return self._content_url
+
+    @property
+    def created_at(self) -> Optional["datetime"]:
+        return self._created_at
+
+    @property
+    def id(self) -> Optional[str]:
+        return self._id
+
+    @property
+    def image(self) -> bytes:
+        if self._image is None:
+            error = "View item must be populated with its png image first."
+            raise UnpopulatedPropertyError(error)
+        return self._image()
+
+    @property
+    def name(self) -> Optional[str]:
+        return self._name
+
+    @name.setter
+    def name(self, value: str):
+        self._name = value
+
+    @property
+    def shared(self) -> Optional[bool]:
+        return self._shared
+
+    @shared.setter
+    def shared(self, value: bool):
+        self._shared = value
+
+    @property
+    def updated_at(self) -> Optional["datetime"]:
+        return self._updated_at
+
+    @property
+    def owner(self) -> Optional[UserItem]:
+        return self._owner
+
+    @owner.setter
+    def owner(self, value: UserItem):
+        self._owner = value
+
+    @property
+    def workbook(self) -> Optional[WorkbookItem]:
+        return self._workbook
+
+    @property
+    def view(self) -> Optional[ViewItem]:
+        return self._view
+
+    @classmethod
+    def from_response(cls, resp, ns, workbook_id="") -> Optional["CustomViewItem"]:
+        item = cls.list_from_response(resp, ns, workbook_id)
+        if not item or len(item) == 0:
+            return None
+        else:
+            return item[0]
+
+    @classmethod
+    def list_from_response(cls, resp, ns, workbook_id="") -> List["CustomViewItem"]:
+        return cls.from_xml_element(fromstring(resp), ns, workbook_id)
+
+    """
+    <customView
+    id="37d015c6-bc28-4c88-989c-72c0a171f7aa"
+    name="New name 2"
+    createdAt="2016-02-03T23:35:09Z"
+    updatedAt="2022-09-28T23:56:01Z"
+    shared="false">
+      <view id="8e33ff19-a7a4-4aa5-9dd8-a171e2b9c29f" name="circle"/>
+      <workbook id="2fbe87c9-a7d8-45bf-b2b3-877a26ec9af5" name="marks and viz types 2"/>
+      <owner id="cdfe8548-84c8-418e-9b33-2c0728b2398a" name="workgroupuser"/>
+    </customView>
+    """
+
+    @classmethod
+    def from_xml_element(cls, parsed_response, ns, workbook_id="") -> List["CustomViewItem"]:
+        all_view_items = list()
+        all_view_xml = parsed_response.findall(".//t:customView", namespaces=ns)
+        for custom_view_xml in all_view_xml:
+            cv_item = cls()
+            view_elem: ElementTree = custom_view_xml.find(".//t:view", namespaces=ns)
+            workbook_elem: str = custom_view_xml.find(".//t:workbook", namespaces=ns)
+            owner_elem: str = custom_view_xml.find(".//t:owner", namespaces=ns)
+            cv_item._created_at = parse_datetime(custom_view_xml.get("createdAt", None))
+            cv_item._updated_at = parse_datetime(custom_view_xml.get("updatedAt", None))
+            cv_item._content_url = custom_view_xml.get("contentUrl", None)
+            cv_item._id = custom_view_xml.get("id", None)
+            cv_item._name = custom_view_xml.get("name", None)
+
+            if owner_elem is not None:
+                parsed_owners = UserItem.from_response_as_owner(tostring(custom_view_xml), ns)
+                if parsed_owners and len(parsed_owners) > 0:
+                    cv_item._owner = parsed_owners[0]
+
+            if view_elem is not None:
+                parsed_views = ViewItem.from_response(tostring(custom_view_xml), ns)
+                if parsed_views and len(parsed_views) > 0:
+                    cv_item._view = parsed_views[0]
+
+            if workbook_id:
+                cv_item._workbook = WorkbookItem(workbook_id)
+            elif workbook_elem is not None:
+                parsed_workbooks = WorkbookItem.from_response(tostring(custom_view_xml), ns)
+                if parsed_workbooks and len(parsed_workbooks) > 0:
+                    cv_item._workbook = parsed_workbooks[0]
+
+            all_view_items.append(cv_item)
+        return all_view_items

tableauserverclient/models/user_item.py

@@ -92,6 +92,10 @@ def external_auth_user_id(self) -> Optional[str]:
     def id(self) -> Optional[str]:
         return self._id
 
+    @id.setter
+    def id(self, value: str) -> None:
+        self._id = value
+
     @property
     def last_login(self) -> Optional[datetime]:
         return self._last_login
@@ -101,7 +105,6 @@ def name(self) -> Optional[str]:
         return self._name
 
     @name.setter
-    @property_not_empty
     def name(self, value: str):
         self._name = value
 
@@ -205,9 +208,19 @@ def _set_values(
 
     @classmethod
     def from_response(cls, resp, ns) -> List["UserItem"]:
+        element_name = ".//t:user"
+        return cls._parse_xml(element_name, resp, ns)
+
+    @classmethod
+    def from_response_as_owner(cls, resp, ns) -> List["UserItem"]:
+        element_name = ".//t:owner"
+        return cls._parse_xml(element_name, resp, ns)
+
+    @classmethod
+    def _parse_xml(cls, element_name, resp, ns):
         all_user_items = []
         parsed_response = fromstring(resp)
-        all_user_xml = parsed_response.findall(".//t:user", namespaces=ns)
+        all_user_xml = parsed_response.findall(element_name, namespaces=ns)
         for user_xml in all_user_xml:
             (
                 id,

tableauserverclient/server/endpoint/__init__.py

@@ -1,4 +1,5 @@
 from .auth_endpoint import Auth
+from .custom_views_endpoint import CustomViews
 from .data_acceleration_report_endpoint import DataAccelerationReport
 from .data_alert_endpoint import DataAlerts
 from .databases_endpoint import Databases

tableauserverclient/server/endpoint/custom_views_endpoint.py

@@ -0,0 +1,104 @@
+import logging
+from typing import List, Optional, Tuple
+
+from .endpoint import QuerysetEndpoint, api
+from .exceptions import MissingRequiredFieldError
+from tableauserverclient.models import CustomViewItem, PaginationItem
+from tableauserverclient.server import RequestFactory, RequestOptions, ImageRequestOptions
+
+logger = logging.getLogger("tableau.endpoint.custom_views")
+
+"""
+Get a list of custom views on a site
+get the details of a custom view
+download an image of a custom view.
+Delete a custom view 
+update the name or owner of a custom view.
+"""
+
+
+class CustomViews(QuerysetEndpoint):
+    def __init__(self, parent_srv):
+        super(CustomViews, self).__init__(parent_srv)
+
+    @property
+    def baseurl(self) -> str:
+        return "{0}/sites/{1}/customviews".format(self.parent_srv.baseurl, self.parent_srv.site_id)
+
+    """
+    If the request has no filter parameters: Administrators will see all custom views. 
+        Other users will see only custom views that they own.
+    If the filter parameters include ownerId: Users will see only custom views that they own.
+    If the filter parameters include viewId and/or workbookId, and don't include ownerId:
+        Users will see those custom views that they have Write and WebAuthoring permissions for.
+    If site user visibility is not set to Limited, the Users will see those custom views that are "public",
+     meaning the value of their shared attribute is true.
+    If site user visibility is set to Limited, ????
+    """
+
+    @api(version="3.18")
+    def get(self, req_options: Optional["RequestOptions"] = None) -> Tuple[List[CustomViewItem], PaginationItem]:
+        logger.info("Querying all custom views 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_view_items = CustomViewItem.list_from_response(server_response.content, self.parent_srv.namespace)
+        return all_view_items, pagination_item
+
+    @api(version="3.18")
+    def get_by_id(self, view_id: str) -> Optional[CustomViewItem]:
+        if not view_id:
+            error = "Custom view item missing ID."
+            raise MissingRequiredFieldError(error)
+        logger.info("Querying custom view (ID: {0})".format(view_id))
+        url = "{0}/{1}".format(self.baseurl, view_id)
+        server_response = self.get_request(url)
+        return CustomViewItem.from_response(server_response.content, self.parent_srv.namespace)
+
+    @api(version="3.18")
+    def populate_image(self, view_item: CustomViewItem, req_options: Optional["ImageRequestOptions"] = None) -> None:
+        if not view_item.id:
+            error = "Custom View item missing ID."
+            raise MissingRequiredFieldError(error)
+
+        def image_fetcher():
+            return self._get_view_image(view_item, req_options)
+
+        view_item._set_image(image_fetcher)
+        logger.info("Populated image for custom view (ID: {0})".format(view_item.id))
+
+    def _get_view_image(self, view_item: CustomViewItem, req_options: Optional["ImageRequestOptions"]) -> bytes:
+        url = "{0}/{1}/image".format(self.baseurl, view_item.id)
+        server_response = self.get_request(url, req_options)
+        image = server_response.content
+        return image
+
+    """
+    Not yet implemented: pdf or csv exports
+    """
+
+    @api(version="3.18")
+    def update(self, view_item: CustomViewItem) -> Optional[CustomViewItem]:
+        if not view_item.id:
+            error = "Custom view item missing ID."
+            raise MissingRequiredFieldError(error)
+        if not (view_item.owner or view_item.name or view_item.shared):
+            logger.debug("No changes to make")
+            return view_item
+
+        # Update the custom view owner or name
+        url = "{0}/{1}".format(self.baseurl, view_item.id)
+        update_req = RequestFactory.CustomView.update_req(view_item)
+        server_response = self.put_request(url, update_req)
+        logger.info("Updated custom view (ID: {0})".format(view_item.id))
+        return CustomViewItem.from_response(server_response.content, self.parent_srv.namespace)
+
+    # Delete 1 view by id
+    @api(version="3.19")
+    def delete(self, view_id: str) -> None:
+        if not view_id:
+            error = "Custom View ID undefined."
+            raise ValueError(error)
+        url = "{0}/{1}".format(self.baseurl, view_id)
+        self.delete_request(url)
+        logger.info("Deleted single custom view (ID: {0})".format(view_id))

tableauserverclient/server/request_factory.py

@@ -1127,10 +1127,21 @@ def update_req(self, xml_request: ET.Element, metric_item: MetricItem) -> bytes:
         return ET.tostring(xml_request)
 
 
+class CustomViewRequest(object):
+    @_tsrequest_wrapped
+    def update_req(self, xml_request: ET.Element, custom_view_item: CustomViewItem):
+        updating_element = ET.SubElement(xml_request, "customView")
+        if custom_view_item.owner is not None and custom_view_item.owner.id is not None:
+            ET.SubElement(updating_element, "owner", {"id": custom_view_item.owner.id})
+        if custom_view_item.name is not None:
+            updating_element.attrib["name"] = custom_view_item.name
+
+
 class RequestFactory(object):
     Auth = AuthRequest()
     Connection = Connection()
     Column = ColumnRequest()
+    CustomView = CustomViewRequest()
     DataAlert = DataAlertRequest()
     Datasource = DatasourceRequest()
     Database = DatabaseRequest()