8000 Jorwoods/type hint groups by jorwoods · Pull Request #938 · tableau/server-client-python · GitHub
[go: up one dir, main page]
Skip to content

Jorwoods/type hint groups #938

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Jan 28, 2022
Merged

Jorwoods/type hint groups #938

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
8a342c2
Type hint groups
jorwoods Nov 17, 2021
5432500
Type hint GroupRequest
jorwoods Oct 24, 2021
97158a4
Merge branch 'development' into jorwoods/type_hint_groups
jacalata Jan 28, 2022
4a27a06
Merge branch 'development' into jorwoods/type_hint_groups
jacalata Jan 28, 2022
be29ed9
Merge branch 'development' into jorwoods/type_hint_groups
jacalata Jan 28, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 29 additions & 24 deletions tableauserverclient/models/group_item.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,78 +4,83 @@
from .reference_item import ResourceReference
from .user_item import UserItem

from typing import Callable, List, Optional, TYPE_CHECKING

if TYPE_CHECKING:
from ..server import Pager


class GroupItem(object):

tag_name = "group"
tag_name: str = "group"

class LicenseMode:
onLogin = "onLogin"
onSync = "onSync"
onLogin: str = "onLogin"
onSync: str = "onSync"

def __init__(self, name=None, domain_name=None):
self._id = None
self._license_mode = None
self._minimum_site_role = None
self._users = None
self.name = name
self.domain_name = domain_name
def __init__(self, name=None, domain_name=None) -> None:
self._id: Optional[str] = None
self._license_mode: Optional[str] = None
self._minimum_site_role: Optional[str] = None
self._users: Optional[Callable[..., "Pager"]] = None
self.name: Optional[str] = name
self.domain_name: Optional[str] = domain_name

@property
def domain_name(self):
def domain_name(self) -> Optional[str]:
return self._domain_name

@domain_name.setter
def domain_name(self, value):
def domain_name(self, value: str) -> None:
self._domain_name = value

@property
def id(self):
def id(self) -> Optional[str]:
return self._id

@property
def name(self):
def name(self) -> Optional[str]:
return self._name

@name.setter
@property_not_empty
def name(self, value):
def name(self, value: str) -> None:
self._name = value

@property
def license_mode(self):
def license_mode(self) -> Optional[str]:
return self._license_mode

@license_mode.setter
@property_is_enum(LicenseMode)
def license_mode(self, value):
def license_mode(self, value: str) -> None:
self._license_mode = value

@property
def minimum_site_role(self):
def minimum_site_role(self) -> Optional[str]:
return self._minimum_site_role

@minimum_site_role.setter
@property_is_enum(UserItem.Roles)
def minimum_site_role(self, value):
def minimum_site_role(self, value: str) -> None:
self._minimum_site_role = value

@property
def users(self):
def users(self) -> "Pager":
if self._users is None:
error = "Group must be populated with users first."
raise UnpopulatedPropertyError(error)
# Each call to `.users` should create a new pager, this just runs the callable
return self._users()

def to_reference(self):
def to_reference(self) -> ResourceReference:
return ResourceReference(id_=self.id, tag_name=self.tag_name)

def _set_users(self, users):
def _set_users(self, users: Callable[..., "Pager"]) -> None:
self._users = users

@classmethod
def from_response(cls, resp, ns):
def from_response(cls, resp, ns) -> List["GroupItem"]:
all_group_items = list()
parsed_response = ET.fromstring(resp)
all_group_xml = parsed_response.findall(".//t:group", namespaces=ns)
Expand All @@ -100,5 +105,5 @@ def from_response(cls, resp, ns):
return all_group_items

@staticmethod
def as_reference(id_):
def as_reference(id_: str) -> ResourceReference:
return ResourceReference(id_, GroupItem.tag_name)
29 changes: 19 additions & 10 deletions tableauserverclient/server/endpoint/groups_endpoint.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,15 +7,20 @@

logger = logging.getLogger("tableau.endpoint.groups")

from typing import List, Optional, TYPE_CHECKING, Tuple, Union

if TYPE_CHECKING:
from ..request_options import RequestOptions


class Groups(Endpoint):
@property
def baseurl(self):
def baseurl(self) -> str:
return "{0}/sites/{1}/groups".format(self.parent_srv.baseurl, self.parent_srv.site_id)

# Gets all groups
@api(version="2.0")
def get(self, req_options=None):
def get(self, req_options: Optional["RequestOptions"] = None) -> Tuple[List[GroupItem], PaginationItem]:
logger.info("Querying all groups on site")
url = self.baseurl
server_response = self.get_request(url, req_options)
Expand All @@ -25,7 +30,7 @@ def get(self, req_options=None):

# Gets all users in a given group
@api(version="2.0")
def populate_users(self, group_item, req_options=None):
def populate_users(self, group_item, req_options: Optional["RequestOptions"] = None) -> None:
if not group_item.id:
error = "Group item missing ID. Group must be retrieved from server first."
raise MissingRequiredFieldError(error)
Expand All @@ -40,7 +45,9 @@ def user_pager():

group_item._set_users(user_pager)

def _get_users_for_group(self, group_item, req_options=None):
def _get_users_for_group(
self, group_item, req_options: Optional["RequestOptions"] = None
) -> Tuple[List[GroupItem], 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)
Expand All @@ -50,7 +57,7 @@ def _get_users_for_group(self, group_item, req_options=None):

# Deletes 1 group by id
@api(version="2.0")
def delete(self, group_id):
def delete(self, group_id: str) -> None:
if not group_id:
error = "Group ID undefined."
raise ValueError(error)
Expand All @@ -59,7 +66,9 @@ def delete(self, group_id):
logger.info("Deleted single group (ID: {0})".format(group_id))

@api(version="2.0")
def update(self, group_item, default_site_role=None, as_job=False):
def update(
self, group_item: GroupItem, default_site_role: Optional[str] = None, as_job: bool = False
) -> Union[GroupItem, JobItem]:
# (1/8/2021): Deprecated starting v0.15
if default_site_role is not None:
import warnings
Expand Down Expand Up @@ -90,15 +99,15 @@ def update(self, group_item, default_site_role=None, as_job=False):

# Create a 'local' Tableau group
@api(version="2.0")
def create(self, group_item):
def create(self, group_item: GroupItem) -> GroupItem:
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, sel 8000 f.parent_srv.namespace)[0]

# Create a group based on Active Directory
@api(version="2.0")
def create_AD_group(self, group_item, asJob=False):
def create_AD_group(self, group_item: GroupItem, asJob: bool = False) -> Union[GroupItem, JobItem]:
asJobparameter = "?asJob=true" if asJob else ""
url = self.baseurl + asJobparameter
create_req = RequestFactory.Group.create_ad_req(group_item)
Expand All @@ -110,7 +119,7 @@ def create_AD_group(self, group_item, asJob=False):

# Removes 1 user from 1 group
@api(version="2.0")
def remove_user(self, group_item, user_id):
def remove_user(self, group_item: GroupItem, user_id: str) -> None:
if not group_item.id:
error = "Group item missing ID."
raise MissingRequiredFieldError(error)
Expand All @@ -123,7 +132,7 @@ def remove_user(self, group_item, user_id):

# Adds 1 user to 1 group
@api(version="2.0")
def add_user(self, group_item, user_id):
def add_user(self, group_item: GroupItem, user_id: str) -> UserItem:
if not group_item.id:
error = "Group item missing ID."
raise MissingRequiredFieldError(error)
Expand Down
30 changes: 21 additions & 9 deletions tableauserverclient/server/request_factory.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,6 @@
from ..models import FlowItem
from ..models import ConnectionItem


def _add_multipart(parts: Dict) -> Tuple[Any, str]:
mime_multipart_parts = list()
for name, (filename, data, content_type) in parts.items():
Expand Down Expand Up @@ -353,24 +352,30 @@ def publish_req_chunked(self, flow_item, connections=None) -> Tuple[Any, str]:


class GroupRequest(object):
def add_user_req(self, user_id):
def add_user_req(self, user_id: str) -> bytes:
xml_request = ET.Element("tsRequest")
user_element = ET.SubElement(xml_request, "user")
user_element.attrib["id"] = user_id
return ET.tostring(xml_request)

def create_local_req(self, group_item):
def create_local_req(self, group_item: GroupItem) -> bytes:
xml_request = ET.Element("tsRequest")
group_element = ET.SubElement(xml_request, "group")
group_element.attrib["name"] = group_item.name
if group_item.name is not None:
group_element.attrib["name"] = group_item.name
else:
raise ValueError("Group name must be populated")
if group_item.minimum_site_role is not None:
group_element.attrib["minimumSiteRole"] = group_item.minimum_site_role
return ET.tostring(xml_request)

def create_ad_req(self, group_item):
def create_ad_req(self, group_item: GroupItem) -> bytes:
xml_request = ET.Element("tsRequest")
group_element = ET.SubElement(xml_request, "group")
group_element.attrib["name"] = group_item.name
if group_item.name is not None:
group_element.attrib["name"] = group_item.name
else:
raise ValueError("Group name must be populated")
import_element = ET.SubElement(group_element, "import")
import_element.attrib["source"] = "ActiveDirectory"
if group_item.domain_name is None:
Expand All @@ -384,7 +389,7 @@ def create_ad_req(self, group_item):
import_element.attrib["siteRole"] = group_item.minimum_site_role
return ET.tostring(xml_request)

def update_req(self, group_item, default_site_role=None):
def update_req(self, group_item: GroupItem, default_site_role: Optional[str] = None) -> bytes:
# (1/8/2021): Deprecated starting v0.15
if default_site_role is not None:
import warnings
Expand All @@ -399,13 +404,20 @@ def update_req(self, group_item, default_site_role=None):

xml_request = ET.Element("tsRequest")
group_element = ET.SubElement(xml_request, "group")
group_element.attrib["name"] = group_item.name

if group_item.name is not None:
group_element.attrib["name"] = group_item.name
else:
raise ValueError("Group name must be populated")
if group_item.domain_name is not None and group_item.domain_name != "local":
# Import element is only accepted in the request for AD groups
import_element = ET.SubElement(group_element, "import")
import_element.attrib["source"] = "ActiveDirectory"
import_element.attrib["domainName"] = group_item.domain_name
import_element.attrib["siteRole"] = group_item.minimum_site_role
if isinstance(group_item.minimum_site_role, str):
import_element.attrib["siteRole"] = group_item.minimum_site_role
else:
raise ValueError("Minimum site role must be provided.")
if group_item.license_mode is not None:
import_element.attrib["grantLicenseMode"] = group_item.license_mode
else:
Expand Down
Loading
0