Skip to content

Commit

Permalink
Add admin commands (#30)
Browse files Browse the repository at this point in the history
* Admin add/remove a user from a group

* Admin Enable/disable a user

* Fixed command and added comments

* Added listing group names a user belongs to

* Fixed style  via Black
  • Loading branch information
circulon authored Feb 17, 2021
1 parent fe99470 commit 3e89204
Showing 1 changed file with 78 additions and 0 deletions.
78 changes: 78 additions & 0 deletions pycognito/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -729,3 +729,81 @@ def get_groups(self):
"""
response = self.client.list_groups(UserPoolId=self.user_pool_id)
return [self.get_group_obj(group_data) for group_data in response.get("Groups")]

def admin_add_user_to_group(self, username, group_name):
"""
Add the user to the specified group
:param username: the username
:param group_name: the name of the group to add the user to
:return:
"""
self.client.admin_add_user_to_group(
UserPoolId=self.user_pool_id,
Username=username,
GroupName=group_name,
)

def admin_remove_user_from_group(self, username, group_name):
"""
Remove the user from the specified group
:param username: the username
:param group_name: the name of the group to remove the user from
:return:
"""
self.client.admin_remove_user_from_group(
UserPoolId=self.user_pool_id,
Username=username,
GroupName=group_name,
)

def admin_list_groups_for_user(self, username):
"""
Get the list of groups a user belongs to
:param username:
:return: List
"""

def process_groups_response(groups_response):
groups = []
for group_dict in groups_response["Groups"]:
groups.append(group_dict["GroupName"])
return groups

groups_response = self.client.admin_list_groups_for_user(
Username=username, UserPoolId=self.user_pool_id, Limit=60
)
user_groups = process_groups_response(groups_response)

while "NextToken" in groups_response.keys():
groups_response = self.client.admin_list_groups_for_user(
Username=username,
UserPoolId=self.user_pool_id,
Limit=60,
NextToken=groups_response["NextToken"],
)
new_groups = process_groups_response(groups_response)
user_groups.extend(new_groups)

return user_groups

def admin_enable_user(self, username):
"""
Enable a user
:param username:
:return:
"""
self.client.admin_enable_user(
UserPoolId=self.user_pool_id,
Username=username,
)

def admin_disable_user(self, username):
"""
Disable a user
:param username:
:return:
"""
self.client.admin_disable_user(
UserPoolId=self.user_pool_id,
Username=self.username,
)

0 comments on commit 3e89204

Please sign in to comment.