feat: custom views dl pub

tableauserverclient/server/endpoint/custom_views_endpoint.py

@@ -1,8 +1,13 @@
+import io
 import logging
-from typing import List, Optional, Tuple
-
-from .endpoint import QuerysetEndpoint, api
-from .exceptions import MissingRequiredFieldError
+import os
+from pathlib import Path
+from typing import List, Optional, Tuple, Union
+
+from tableauserverclient.config import BYTES_PER_MB, FILESIZE_LIMIT_MB
+from tableauserverclient.filesys_helpers import get_file_object_size
+from tableauserverclient.server.endpoint.endpoint import QuerysetEndpoint, api
+from tableauserverclient.server.endpoint.exceptions import MissingRequiredFieldError
 from tableauserverclient.models import CustomViewItem, PaginationItem
 from tableauserverclient.server import RequestFactory, RequestOptions, ImageRequestOptions
 
@@ -16,6 +21,15 @@
 update the name or owner of a custom view.
 """
 
+FilePath = Union[str, os.PathLike]
+FileObject = Union[io.BufferedReader, io.BytesIO]
+FileObjectR = Union[io.BufferedReader, io.BytesIO]
+FileObjectW = Union[io.BufferedWriter, io.BytesIO]
+PathOrFileR = Union[FilePath, FileObjectR]
+PathOrFileW = Union[FilePath, FileObjectW]
+io_types_r = (io.BufferedReader, io.BytesIO)
+io_types_w = (io.BufferedWriter, io.BytesIO)
+
 
 class CustomViews(QuerysetEndpoint[CustomViewItem]):
     def __init__(self, parent_srv):
@@ -25,6 +39,10 @@ def __init__(self, parent_srv):
     def baseurl(self) -> str:
         return "{0}/sites/{1}/customviews".format(self.parent_srv.baseurl, self.parent_srv.site_id)
 
+    @property
+    def expurl(self) -> str:
+        return f"{self.parent_srv._server_address}/api/exp/sites/{self.parent_srv.site_id}/customviews"
+
     """
     If the request has no filter parameters: Administrators will see all custom views. 
         Other users will see only custom views that they own.
@@ -102,3 +120,46 @@ def delete(self, view_id: str) -> None:
         url = "{0}/{1}".format(self.baseurl, view_id)
         self.delete_request(url)
         logger.info("Deleted single custom view (ID: {0})".format(view_id))
+
+    @api(version="3.21")
+    def download(self, view_item: CustomViewItem, file: PathOrFileW) -> PathOrFileW:
+        url = f"{self.expurl}/{view_item.id}/content"
+        server_response = self.get_request(url)
+        if isinstance(file, io_types_w):
+            file.write(server_response.content)
+            return file
+
+        with open(file, "wb") as f:
+            f.write(server_response.content)
+
+        return file
+
+    @api(version="3.21")
+    def publish(self, view_item: CustomViewItem, file: PathOrFileR) -> Optional[CustomViewItem]:
+        url = self.expurl
+        if isinstance(file, io_types_r):
+            size = get_file_object_size(file)
+        elif isinstance(file, (str, Path)) and (p := Path(file)).is_file():
+            size = p.stat().st_size
+        else:
+            raise ValueError("File path or file object required for publishing custom view.")
+
+        if size >= FILESIZE_LIMIT_MB * BYTES_PER_MB:
+            upload_session_id = self.parent_srv.fileuploads.upload(file)
+            url = f"{url}?uploadSessionId={upload_session_id}"
+            xml_request, content_type = RequestFactory.CustomView.publish_req_chunked(view_item)
+        else:
+            if isinstance(file, io_types_r):
+                file.seek(0)
+                contents = file.read()
+                if view_item.name is None:
+                    raise MissingRequiredFieldError("Custom view item missing name.")
+                filename = view_item.name
+            elif isinstance(file, (str, Path)):
+                filename = Path(file).name
+                contents = Path(file).read_bytes()
+
+            xml_request, content_type = RequestFactory.CustomView.publish_req(view_item, filename, contents)
+
+        server_response = self.post_request(url, xml_request, content_type)
+        return CustomViewItem.from_response(server_response.content, self.parent_srv.namespace)

tableauserverclient/server/request_factory.py

@@ -1,4 +1,5 @@
 import xml.etree.ElementTree as ET
+
 from typing import Any, Dict, Iterable, List, Optional, Tuple, TYPE_CHECKING, Union
 
 from requests.packages.urllib3.fields import RequestField
@@ -1267,6 +1268,49 @@ def update_req(self, xml_request: ET.Element, custom_view_item: CustomViewItem):
         if custom_view_item.name is not None:
             updating_element.attrib["name"] = custom_view_item.name
 
+    @_tsrequest_wrapped
+    def _publish_xml(self, xml_request: ET.Element, custom_view_item: CustomViewItem) -> bytes:
+        custom_view_element = ET.SubElement(xml_request, "customView")
+        if (name := custom_view_item.name) is not None:
+            custom_view_element.attrib["name"] = name
+        else:
+            raise ValueError(f"Custom View Item missing name: {custom_view_item}")
+        if (shared := custom_view_item.shared) is not None:
+            custom_view_element.attrib["shared"] = str(shared).lower()
+        else:
+            raise ValueError(f"Custom View Item missing shared: {custom_view_item}")
+        if (owner := custom_view_item.owner) is not None:
+            owner_element = ET.SubElement(custom_view_element, "owner")
+            if (owner_id := owner.id) is not None:
+                owner_element.attrib["id"] = owner_id
+            else:
+                raise ValueError(f"Custom View Item owner missing id: {owner}")
+        else:
+            raise ValueError(f"Custom View Item missing owner: {custom_view_item}")
+        if (workbook := custom_view_item.workbook) is not None:
+            workbook_element = ET.SubElement(custom_view_element, "workbook")
+            if (workbook_id := workbook.id) is not None:
+                workbook_element.attrib["id"] = workbook_id
+            else:
+                raise ValueError(f"Custom View Item workbook missing id: {workbook}")
+        else:
+            raise ValueError(f"Custom View Item missing workbook: {custom_view_item}")
+
+        return ET.tostring(xml_request)
+
+    def publish_req_chunked(self, custom_view_item: CustomViewItem):
+        xml_request = self._publish_xml(custom_view_item)
+        parts = {"request_payload": ("", xml_request, "text/xml")}
+        return _add_multipart(parts)
+
+    def publish_req(self, custom_view_item: CustomViewItem, filename: str, file_contents: bytes):
+        xml_request = self._publish_xml(custom_view_item)
+        parts = {
+            "request_payload": ("", xml_request, "text/xml"),
+            "tableau_customview": (filename, file_contents, "application/octet-stream"),
+        }
+        return _add_multipart(parts)
+
 
 class GroupSetRequest:
     @_tsrequest_wrapped