-
Notifications
You must be signed in to change notification settings - Fork 421
/
custom_views_endpoint.py
165 lines (139 loc) · 7.06 KB
/
custom_views_endpoint.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
import io
import logging
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
from tableauserverclient.helpers.logging import logger
"""
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.
"""
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):
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)
@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.
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))
@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)