forked from tableau/server-client-python
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathgroups_endpoint.py
164 lines (140 loc) · 7.5 KB
/
groups_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
import logging
from tableauserverclient.server.endpoint.endpoint import QuerysetEndpoint, api
from tableauserverclient.server.endpoint.exceptions import MissingRequiredFieldError
from tableauserverclient.server import RequestFactory
from tableauserverclient.models import GroupItem, UserItem, PaginationItem, JobItem
from tableauserverclient.server.pager import Pager
from tableauserverclient.helpers.logging import logger
from typing import Iterable, List, Optional, TYPE_CHECKING, Tuple, Union
if TYPE_CHECKING:
from tableauserverclient.server.request_options import RequestOptions
class Groups(QuerysetEndpoint[GroupItem]):
@property
def baseurl(self) -> str:
return "{0}/sites/{1}/groups".format(self.parent_srv.baseurl, self.parent_srv.site_id)
@api(version="2.0")
def get(self, req_options: Optional["RequestOptions"] = None) -> Tuple[List[GroupItem], PaginationItem]:
"""Gets all groups"""
logger.info("Querying all groups 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_group_items = GroupItem.from_response(server_response.content, self.parent_srv.namespace)
return all_group_items, pagination_item
@api(version="2.0")
def populate_users(self, group_item: GroupItem, req_options: Optional["RequestOptions"] = None) -> None:
"""Gets all users in a given group"""
if not group_item.id:
error = "Group item missing ID. Group must be retrieved from server first."
raise MissingRequiredFieldError(error)
# Define an inner function that we bind to the model_item's `.user` property.
def user_pager():
return Pager(
lambda options: self._get_users_for_group(group_item, options),
req_options,
)
group_item._set_users(user_pager)
def _get_users_for_group(
self, group_item: GroupItem, req_options: Optional["RequestOptions"] = None
) -> Tuple[List[UserItem], PaginationItem]:
url = "{0}/{1}/users".format(self.baseurl, group_item.id)
server_response = self.get_request(url, req_options)
user_item = UserItem.from_response(server_response.content, self.parent_srv.namespace)
pagination_item = PaginationItem.from_response(server_response.content, self.parent_srv.namespace)
logger.info("Populated users for group (ID: {0})".format(group_item.id))
return user_item, pagination_item
@api(version="2.0")
def delete(self, group_id: str) -> None:
"""Deletes 1 group by id"""
if not group_id:
error = "Group ID undefined."
raise ValueError(error)
url = "{0}/{1}".format(self.baseurl, group_id)
self.delete_request(url)
logger.info("Deleted single group (ID: {0})".format(group_id))
@api(version="2.0")
def update(self, group_item: GroupItem, as_job: bool = False) -> Union[GroupItem, JobItem]:
url = "{0}/{1}".format(self.baseurl, group_item.id)
if not group_item.id:
error = "Group item missing ID."
raise MissingRequiredFieldError(error)
if as_job and (group_item.domain_name is None or group_item.domain_name == "local"):
error = "Local groups cannot be updated asynchronously."
raise ValueError(error)
elif as_job:
url = "?".join([url, "asJob=True"])
update_req = RequestFactory.Group.update_req(group_item)
server_response = self.put_request(url, update_req)
logger.info("Updated group item (ID: {0})".format(group_item.id))
if as_job:
return JobItem.from_response(server_response.content, self.parent_srv.namespace)[0]
else:
return GroupItem.from_response(server_response.content, self.parent_srv.namespace)[0]
@api(version="2.0")
def create(self, group_item: GroupItem) -> GroupItem:
"""Create a 'local' Tableau group"""
url = self.baseurl
create_req = RequestFactory.Group.create_local_req(group_item)
server_response = self.post_request(url, create_req)
return GroupItem.from_response(server_response.content, self.parent_srv.namespace)[0]
@api(version="2.0")
def create_AD_group(self, group_item: GroupItem, asJob: bool = False) -> Union[GroupItem, JobItem]:
"""Create a group based on Active Directory"""
asJobparameter = "?asJob=true" if asJob else ""
url = self.baseurl + asJobparameter
create_req = RequestFactory.Group.create_ad_req(group_item)
server_response = self.post_request(url, create_req)
if asJob:
return JobItem.from_response(server_response.content, self.parent_srv.namespace)[0]
else:
return GroupItem.from_response(server_response.content, self.parent_srv.namespace)[0]
@api(version="2.0")
def remove_user(self, group_item: GroupItem, user_id: str) -> None:
"""Removes 1 user from 1 group"""
if not group_item.id:
error = "Group item missing ID."
raise MissingRequiredFieldError(error)
if not user_id:
error = "User ID undefined."
raise ValueError(error)
url = "{0}/{1}/users/{2}".format(self.baseurl, group_item.id, user_id)
self.delete_request(url)
logger.info("Removed user (id: {0}) from group (ID: {1})".format(user_id, group_item.id))
@api(version="3.21")
def remove_users(self, group_item: GroupItem, users: Iterable[Union[str, UserItem]]) -> None:
"""Removes multiple users from 1 group"""
group_id = group_item.id if hasattr(group_item, "id") else group_item
if not isinstance(group_id, str):
raise ValueError(f"Invalid group provided: {group_item}")
url = f"{self.baseurl}/{group_id}/users/remove"
add_req = RequestFactory.Group.remove_users_req(users)
_ = self.put_request(url, add_req)
logger.info("Removed users to group (ID: {0})".format(group_item.id))
return None
@api(version="2.0")
def add_user(self, group_item: GroupItem, user_id: str) -> UserItem:
"""Adds 1 user to 1 group"""
if not group_item.id:
error = "Group item missing ID."
raise MissingRequiredFieldError(error)
if not user_id:
error = "User ID undefined."
raise ValueError(error)
url = "{0}/{1}/users".format(self.baseurl, group_item.id)
add_req = RequestFactory.Group.add_user_req(user_id)
server_response = self.post_request(url, add_req)
user = UserItem.from_response(server_response.content, self.parent_srv.namespace).pop()
logger.info("Added user (id: {0}) to group (ID: {1})".format(user_id, group_item.id))
return user
@api(version="3.21")
def add_users(self, group_item: GroupItem, users: Iterable[Union[str, UserItem]]) -> List[UserItem]:
"""Adds multiple users to 1 group"""
group_id = group_item.id if hasattr(group_item, "id") else group_item
if not isinstance(group_id, str):
raise ValueError(f"Invalid group provided: {group_item}")
url = f"{self.baseurl}/{group_id}/users"
add_req = RequestFactory.Group.add_users_req(users)
server_response = self.post_request(url, add_req)
users = UserItem.from_response(server_response.content, self.parent_srv.namespace)
logger.info("Added users to group (ID: {0})".format(group_item.id))
return users