Skip to content
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

Add some more endpoints for group management #91

Closed
wants to merge 6 commits into from
Closed

Add some more endpoints for group management #91

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ba41074
Implement get_group_statistics()
fschrempf Nov 29, 2023
7b2481e
Implement [create/delete]_group()
fschrempf Nov 29, 2023
e2533a2
Implement get_group_members()
fschrempf Nov 29, 2023
e3ee259
Implement [add/remove]_group_member()
fschrempf Nov 29, 2023
74759bd
Implement get_group_roles()
fschrempf Nov 29, 2023
257b06c
Implement [add/remove]_parent_group()
fschrempf Nov 29, 2023
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
251 changes: 251 additions & 0 deletions churchtools_api/groups.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,75 @@ def get_groups_hierarchies(self):
"Something went wrong fetching groups hierarchies: {}".format(
response.status_code))

def get_group_statistics(self, group_id: int):
"""
Get statistics for the given group
:param group_id: required group_id
:return: dict with statistics
:rtype: dict
"""
url = self.domain + '/api/groups/{}/statistics'.format(group_id)
headers = {
'accept': 'application/json'
}
response = self.session.get(url=url, headers=headers)

if response.status_code == 200:
response_content = json.loads(response.content)
response_data = response_content['data'].copy()
logging.debug(
"First response of Group Statistics successful {}".format(response_content))
return response_data
else:
logging.warning(
"Something went wrong fetching group statistics: {}".format(
response.status_code))

def create_group(self, name: str, group_status_id: int, grouptype_id: int, **kwargs):
"""
Create a new group
:param name: str: required name
:param group_status_id: int: required status id
:param grouptype_id: int: required grouptype id
:keyword campus_id: int: optional campus id
:keyword superior_group_id: int: optional superior group id
:keyword force: bool: set to force create if a group with this name
already exists
:return: dict with created group
:rtype: dict
"""
url = self.domain + '/api/groups'
headers = {
'accept': 'application/json'
}
data = {
"groupStatusId": group_status_id,
"groupTypeId": grouptype_id,
"name": name,
}

if 'campus_id' in kwargs.keys():
data['campusId'] = kwargs['campus_id']

if 'force' in kwargs.keys():
data['force'] = kwargs['force']

if 'superior_group_id' in kwargs.keys():
data['superiorGroupId'] = kwargs['superior_group_id']

response = self.session.post(url=url, headers=headers, data=data)

if response.status_code == 201:
response_content = json.loads(response.content)
response_data = response_content['data'].copy()
logging.debug(
"First response of Create Group successful {}".format(response_content))

return response_data
else:
logging.warning(
"Something went wrong with creating group: {}".format(response.status_code))

def update_group(self, group_id: int, data: dict):
"""
Update a field of the given group
Expand Down Expand Up @@ -110,6 +179,24 @@ def update_group(self, group_id: int, data: dict):
"Something went wrong updating group: {}".format(
response.status_code))

def delete_group(self, group_id: int):
"""
Delete the given group
:param group_id: int: required group_id
:return: True if successful
:rtype: bool
"""
url = self.domain + '/api/groups/{}'.format(group_id)
response = self.session.delete(url=url)

if response.status_code == 204:
logging.debug("First response of Delete Group successful")
return True
else:
logging.warning(
"Something went wrong deleting group: {}".format(
response.status_code))

def get_grouptypes(self, **kwargs):
"""
Get list of all grouptypes
Expand Down Expand Up @@ -164,3 +251,167 @@ def get_group_permissions(self, group_id: int):
logging.warning(
"Something went wrong fetching group permissions: {}".format(
response.status_code))

def get_group_members(self, group_id: int, **kwargs):
"""
Get list of members for the given group
:param group_id: int: required group id
:keyword role_ids: list[int]: optional filter list of role ids
:return: list of group member dicts
:rtype: list[dict]
"""
url = self.domain + '/api/groups/{}/members'.format(group_id)
headers = {
'accept': 'application/json'
}
params = {}

if 'role_ids' in kwargs.keys():
params['role_ids[]'] = kwargs['role_ids']

response = self.session.get(url=url, headers=headers, params=params)

if response.status_code == 200:
response_content = json.loads(response.content)
response_data = response_content['data'].copy()
logging.debug(
"First response of Group Members successful {}".format(response_content))

if 'meta' not in response_content.keys(): # Shortcut without Pagination
return response_data

# Long part extending results with pagination
while response_content['meta']['pagination']['current'] \
< response_content['meta']['pagination']['lastPage']:
logging.info("page {} of {}".format(response_content['meta']['pagination']['current'],
response_content['meta']['pagination']['lastPage']))
params['page'] = response_content['meta']['pagination']['current'] + 1
response = self.session.get(
url=url, headers=headers, params=params)
response_content = json.loads(response.content)
response_data.extend(response_content['data'])

return response_data
else:
logging.warning(
"Something went wrong fetching group members: {}".format(response.status_code))

def add_group_member(self, group_id: int, person_id: int, **kwargs):
"""
Add a member to a group
:param group_id: int: required group id
:param person_id: int: required person id
:keyword grouptype_role_id: int: optional grouptype role id
:keyword group_member_status: str: optional member status
:return: dict with group member
:rtype: dict
"""
url = self.domain + \
'/api/groups/{}/members/{}'.format(group_id, person_id)
headers = {
'accept': 'application/json',
}

data = {}
if 'grouptype_role_id' in kwargs.keys():
data['groupTypeRoleId'] = kwargs['grouptype_role_id']
if 'group_member_status' in kwargs.keys():
data['group_member_status'] = kwargs['group_member_status']

response = self.session.put(
url=url, data=data, headers=headers)

if response.status_code == 200:
response_content = json.loads(response.content)
# For unknown reasons the endpoint returns a list of items instead
# of a single item as specified in the API documentation.
response_data = response_content['data'][0].copy()
logging.debug(
"First response of Add Group Member successful {}".format(response_content))

return response_data
else:
logging.warning(
"Something went wrong adding group member: {}".format(response.status_code))

def remove_group_member(self, group_id: int, person_id: int):
"""
Remove the given group member
:param group_id: int: required group id
:param person_id: int: required person id
:return: True if successful
:rtype: bool
"""
url = self.domain + \
'/api/groups/{}/members/{}'.format(group_id, person_id)
response = self.session.delete(url=url)

if response.status_code == 204:
logging.debug("First response of Remove Group member successful")
return True
else:
logging.warning(
"Something went wrong removing group member: {}".format(
response.status_code))

def get_group_roles(self, group_id: int):
"""
Get list of all roles for the given group
:param group_id: int: required group id
:return: list with group roles dicts
:rtype: list[dict]
"""
url = self.domain + '/api/groups/{}/roles'.format(group_id)
headers = {
'accept': 'application/json'
}
response = self.session.get(url=url, headers=headers)

if response.status_code == 200:
response_content = json.loads(response.content)
response_data = response_content['data'].copy()
logging.debug(
"First response of Group Roles successful {}".format(response_content))

return response_data
else:
logging.warning(
"Something went wrong fetching group roles: {}".format(response.status_code))

def add_parent_group(self, group_id: int, parent_group_id: int):
"""
Add a parent group for a group
:param group_id: int: required group id
:param parent_group_id: int: required parent group id
:return: True if successful
:rtype: bool
"""
url = self.domain + \
'/api/groups/{}/parents/{}'.format(group_id, parent_group_id)
response = self.session.put(url=url)

if response.status_code == 201:
logging.debug("First response of Add Parent Group successful")
return True
else:
logging.warning("Something went wrong adding parent group: {}".format(
response.status_code))

def remove_parent_group(self, group_id: int, parent_group_id: int):
"""
REmove a parent group from a group
:param group_id: int: required group id
:param parent_group_id: int: required parent group id
:return: True if successful
:rtype: bool
"""
url = self.domain + \
'/api/groups/{}/parents/{}'.format(group_id, parent_group_id)
response = self.session.delete(url=url)

if response.status_code == 204:
logging.debug("First response of Remove Parent Group successful")
return True
else:
logging.warning("Something went wrong removing parent group: {}".format(
response.status_code))
Loading
Loading