from six.moves.urllib.parse import quote_plus
from swimlane.core.cache import check_cache
from swimlane.core.cursor import PaginatedCursor
from swimlane.core.resolver import SwimlaneResolver
from swimlane.core.resources.usergroup import Group, User
from swimlane.utils import one_of_keyword_only
[docs]class GroupListCursor(SwimlaneResolver, PaginatedCursor):
"""Handles retrieval and pagination of group list endpoint"""
def __init__(self, swimlane, limit=None):
SwimlaneResolver.__init__(self, swimlane)
PaginatedCursor.__init__(self, limit)
def _parse_raw_element(self, raw_element):
return Group(self._swimlane, raw_element)
def _retrieve_raw_elements(self, page):
response = self._swimlane.request(
'get',
'groups',
params={
'size': self.page_size,
'pageNumber': page
}
)
return response.json().get('items', [])
[docs]class GroupAdapter(SwimlaneResolver):
"""Handles retrieval of Swimlane Group resources"""
[docs] def list(self, limit=None):
"""Retrieve list of all groups
Returns:
:class:`list` of :class:`~swimlane.core.resources.usergroup.Group`: List of all Groups
Raises:
ValueError: If limit is not of type integer or None
"""
if (isinstance(limit, int) and limit > 0) or limit is None:
return GroupListCursor(swimlane=self._swimlane, limit=limit)
raise ValueError('Limit should be a positive whole number greater than 0')
[docs] @check_cache(Group)
@one_of_keyword_only('id', 'name')
def get(self, key, value):
"""Retrieve single group record by id or name
Supports resource cache
Keyword Args:
id (str): Full Group ID
name (str): Group name
Raises:
TypeError: Unexpected or more than one keyword argument provided
ValueError: No matching group found based on provided inputs
ValueError: The lookup value is empty or None
Returns:
Group: Group instance matching provided inputs
"""
if not value:
raise ValueError('The value provided for the key "{0}" cannot be empty or None'.format(key))
if key == 'id':
response = self._swimlane.request('get', 'groups/{}'.format(value))
return Group(self._swimlane, response.json())
else:
response = self._swimlane.request('get', 'groups/lookup?name={}'.format(value))
matched_groups = response.json()
for group_data in matched_groups:
if group_data.get('name') == value:
return Group(self._swimlane, group_data)
raise ValueError('Unable to find group with name "{}"'.format(value))
[docs]class UserListCursor(SwimlaneResolver, PaginatedCursor):
"""Handles retrieval and pagination for user list endpoint"""
def __init__(self, swimlane, limit=None):
SwimlaneResolver.__init__(self, swimlane)
PaginatedCursor.__init__(self, limit)
def _parse_raw_element(self, raw_element):
return User(self._swimlane, raw_element)
def _retrieve_raw_elements(self, page):
response = self._swimlane.request(
'get',
'user',
params={
'size': self.page_size,
'pageNumber': page
}
)
return response.json().get('items', [])
[docs]class UserAdapter(SwimlaneResolver):
"""Handles retrieval of Swimlane User resources"""
[docs] def list(self, limit=None):
"""Retrieve all users
Returns:
:class:`UserListCursor`: Paginated cursor yielding :class:`User` instances
Raises:
ValueError: If limit is not of type integer or None
"""
if (isinstance(limit, int) and limit > 0) or limit is None:
return UserListCursor(swimlane=self._swimlane, limit=limit)
raise ValueError('Limit should be a positive whole number greater than 0')
[docs] @check_cache(User)
@one_of_keyword_only('id', 'display_name')
def get(self, arg, value):
"""Retrieve single user record by id or username
Warnings:
User display names are not unique. If using `display_name`, method will fail if multiple Users are returned
with the same display name
Keyword Args:
id (str): Full User ID
display_name (str): User display name
Returns:
User: User instance matching provided inputs
Raises:
TypeError: Unexpected or more than one keyword argument provided
ValueError: No matching user found based on provided inputs, or multiple Users with same display name
ValueError: The lookup value is empty or None
"""
if not value:
raise ValueError('The value provided for the key "{0}" cannot be empty or None'.format(arg))
if arg == 'id':
response = self._swimlane.request('get', 'user/{}'.format(value))
try:
user_data = response.json()
except ValueError:
raise ValueError('Unable to find user with ID "{}"'.format(value))
return User(self._swimlane, user_data)
else:
response = self._swimlane.request('get', 'user/search?query={}'.format(quote_plus(value)))
matched_users = response.json()
# Display name not unique, fail if multiple users share the same target display name
target_matches = []
for user_data in matched_users:
user_display_name = user_data.get('displayName')
if user_display_name == value:
target_matches.append(user_data)
# No matches
if not target_matches:
raise ValueError('Unable to find user with display name "{}"'.format(value))
# Multiple matches
if len(target_matches) > 1:
raise ValueError('Multiple users returned with display name "{}". Matching user IDs: {}'.format(
value,
', '.join(['"{}"'.format(r['id']) for r in target_matches])
))
return User(self._swimlane, target_matches[0])
