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.

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 list_user_pool_clients with pagination and better pagination support for users and groups #267

Closed
wants to merge 3 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
84 changes: 78 additions & 6 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -99,9 +99,9 @@ u = Cognito('your-user-pool-id','your-client-id',

### Examples with Realistic Arguments

#### User Pool Id and Client ID Only
#### User Pool Id

Used when you only need information about the user pool (ex. list users in the user pool)
Used when you only need information about the user pool's clients, groups, or users (ex. list users in the user pool). Client ID can be optionally specified.

```python
from pycognito import Cognito
Expand Down Expand Up @@ -367,19 +367,37 @@ user = u.get_user(attr_map={"given_name":"first_name","family_name":"last_name"}

#### Get Users

Get a list of the user in the user pool.
Get a list of the users in the user pool.

```python
from pycognito import Cognito

u = Cognito('your-user-pool-id','your-client-id')
u = Cognito('your-user-pool-id', 'your-client-id')

user = u.get_users(attr_map={"given_name":"first_name","family_name":"last_name"})
```

You can paginate through retrieving users by specifying the page_limit and page_token arguments.

```python
from pycognito import Cognito

u = Cognito('your-user-pool-id', 'your-client-id')

users = u.get_users(page_limit=10)
page_token = u.get_users_pagination_token()
while page_token:
more_users = u.get_users(page_limit=10, page_token=page_token)
users.extend(more_users)
page_token = u.get_users_pagination_token()
```

##### Arguments

- **attr_map:** Dictionary map from Cognito attributes to attribute names we would like to show to our users
- **pool_id:** The user pool ID to list clients for (uses self.user_pool_id if None)
- **page_limit:** Max results to return from this request (0 to 60)
- **page_token:** Used to return the next set of items

#### Get Group object

Expand Down Expand Up @@ -417,16 +435,70 @@ group = u.get_group(group_name='some_group_name')

#### Get Groups

Get a list of groups in the user pool. Requires developer credentials.
Get a list of groups in the specified user pool (defaults to user pool set on instantiation if not specified). Requires developer credentials.

```python
from pycognito import Cognito

u = Cognito('your-user-pool-id','your-client-id')
u = Cognito('your-user-pool-id', 'your-client-id')

groups = u.get_groups()
```

You can paginate through retrieving groups by specifying the page_limit and page_token arguments.

```python
from pycognito import Cognito

u = Cognito('your-user-pool-id', 'your-client-id')

groups = u.get_groups(page_limit=10)
page_token = u.get_groups_pagination_token()
while page_token:
more_groups = u.get_groups(page_limit=10, page_token=page_token)
groups.extend(more_groups)
page_token = u.get_groups_pagination_token()
```

##### Arguments

- **pool_id:** The user pool ID to list groups for (uses self.user_pool_id if None)
- **page_limit:** Max results to return from this request (0 to 60)
- **page_token:** Used to return the next set of items

#### List User Pool Clients

Returns a list of client dicts of the specified user pool (defaults to user pool set on instantiation if not specified). Requires developer credentials.

```python
from pycognito import Cognito

u = Cognito('your-user-pool-id', 'your-client-id')

clients = u.list_user_pool_clients()
```

You can paginate through retrieving clients by specifying the page_limit and page_token arguments.

```python
from pycognito import Cognito

u = Cognito('your-user-pool-id', 'your-client-id')

clients = u.list_user_pool_clients(page_limit=10)
page_token = u.get_clients_pagination_token()
while page_token:
more_clients = u.list_user_pool_clients(page_limit=10, page_token=page_token)
clients.extend(more_clients)
page_token = u.get_clients_pagination_token()
```

##### Arguments

- **pool_id:** The user pool ID to list clients for (uses self.user_pool_id if None)
- **page_limit:** Max results to return from this request (0 to 60)
- **page_token:** Used to return the next set of items

#### Check Token

Checks the exp attribute of the access_token and either refreshes the tokens by calling the renew_access_tokens method or does nothing. **IMPORTANT:** Access token is required
Expand Down
186 changes: 167 additions & 19 deletions pycognito/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,21 @@ def cognito_to_dict(attr_list, attr_map=None):
return attr_dict


def is_cognito_attr_list(attr_list):
"""
:param attr_list: List of User Pool attribute dicts
:return: bool indicating whether the list contains User Pool attribute dicts
"""
if not isinstance(attr_list, list):
return False
for attr_dict in attr_list:
if not isinstance(attr_dict, dict):
return False
if not attr_dict.keys() <= {"Name", "Value"}:
return False
return True


def dict_to_cognito(attributes, attr_map=None):
"""
:param attributes: Dictionary of User Pool attribute names/values
Expand Down Expand Up @@ -146,7 +161,7 @@ class Cognito:
def __init__(
self,
user_pool_id,
client_id,
client_id=None,
user_pool_region=None,
username=None,
id_token=None,
Expand Down Expand Up @@ -208,13 +223,38 @@ def __init__(
else:
self.client = boto3.client("cognito-idp", **boto3_client_kwargs)

self._users_pagination_next_token = None
self._groups_pagination_next_token = None
self._clients_pagination_next_token = None

@property
def user_pool_url(self):
if self.pool_domain_url:
return f"{self.pool_domain_url}/{self.user_pool_id}"

return f"https://cognito-idp.{self.user_pool_region}.amazonaws.com/{self.user_pool_id}"

def get_users_pagination_token(self) -> str | None:
"""
Returns the pagination token set by the get_users call
:return: str token or None if no more results to request
"""
return self._users_pagination_next_token

def get_groups_pagination_token(self) -> str | None:
"""
Returns the pagination token set by the get_group call
:return: str token or None if no more results to request
"""
return self._groups_pagination_next_token

def get_clients_pagination_token(self) -> str | None:
"""
Returns the pagination token set by the list_user_pool_clients call
:return: str token or None if no more results to request
"""
return self._clients_pagination_next_token

def get_keys(self):
if self.pool_jwk:
return self.pool_jwk
Expand Down Expand Up @@ -534,11 +574,26 @@ def logout(self):
self.access_token = None
self.token_type = None

def admin_update_profile(self, attrs, attr_map=None):
user_attrs = dict_to_cognito(attrs, attr_map)
def admin_update_profile(self, attrs, attr_map=None, username=None):
"""
Updates the user specified (defaults to self.username) with the provided attrs
:param attrs: Dictionary of attribute name, values
:param attr_map: Dictionary map from Cognito attributes to attribute
:param username: Username to update
:return:
"""
if username is None:
username = self.username

# if already formatted for cognito then use as is
if not is_cognito_attr_list(attrs):
user_attrs = dict_to_cognito(attrs, attr_map)
else:
user_attrs = attrs

self.client.admin_update_user_attributes(
UserPoolId=self.user_pool_id,
Username=self.username,
Username=username,
UserAttributes=user_attrs,
)

Expand Down Expand Up @@ -577,25 +632,46 @@ def get_user(self, attr_map=None):
attr_map=attr_map,
)

def get_users(self, attr_map=None):
def get_users(
self,
attr_map=None,
pool_id: str | None = None,
page_limit: int | None = None,
page_token: str | None = None,
) -> list[UserObj]:
"""
Returns all users for a user pool. Returns instances of the
self.user_class.
self.user_class. If page_limit is set then it will return that many (0 to 60)
while setting self._users_pagination_next_token to the next token.
:param attr_map: Dictionary map from Cognito attributes to attribute
names we would like to show to our users
:param pool_id: The user pool ID to list clients for (uses self.user_pool_id if None)
:param page_limit: Max results to return from this request (0 to 60)
:param page_token: Used to return the next set of items
:return: list of self.user_class
"""
response = self.client.list_users(UserPoolId=self.user_pool_id)
if pool_id is None:
pool_id = self.user_pool_id

kwargs = {"UserPoolId": pool_id}
if page_limit:
kwargs["Limit"] = page_limit
if page_token:
kwargs["PaginationToken"] = page_token

response = self.client.list_users(**kwargs)
user_list = response.get("Users")
page_token = response.get("PaginationToken")

while page_token:
response = self.client.list_users(
UserPoolId=self.user_pool_id, PaginationToken=page_token
)
user_list.extend(response.get("Users"))
page_token = response.get("PaginationToken")

if page_limit is None:
while page_token:
response = self.client.list_users(
UserPoolId=pool_id, PaginationToken=page_token
)
user_list.extend(response.get("Users"))
page_token = response.get("PaginationToken")
else:
self._users_pagination_next_token = page_token
return [
self.get_user_obj(
user.get("Username"),
Expand Down Expand Up @@ -805,13 +881,45 @@ def get_group(self, group_name):
)
return self.get_group_obj(response.get("Group"))

def get_groups(self):
"""
Returns all groups for a user pool.
def get_groups(
self,
pool_id: str | None = None,
page_limit: int | None = None,
page_token: str | None = None,
) -> list[GroupObj]:
"""
Returns all groups for a user pool. If page_limit is set then it
will return that many (0 to 60) while setting self._groups_pagination_next_token
to the next token.
:param pool_id: The user pool ID to list clients for (uses self.user_pool_id if None)
:param page_limit: Max results to return from this request (0 to 60)
:param page_token: Used to return the next set of items
:return: list of instances of self.group_class
"""
response = self.client.list_groups(UserPoolId=self.user_pool_id)
return [self.get_group_obj(group_data) for group_data in response.get("Groups")]
if pool_id is None:
pool_id = self.user_pool_id

kwargs = {"UserPoolId": pool_id}
if page_limit:
kwargs["Limit"] = page_limit
if page_token:
kwargs["NextToken"] = page_token

response = self.client.list_groups(**kwargs)
group_list = response.get("Groups")
page_token = response.get("NextToken")

if page_limit is None:
while page_token:
response = self.client.list_groups(
UserPoolId=pool_id, NextToken=page_token
)
group_list.extend(response.get("Groups"))
page_token = response.get("PaginationToken")
else:
self._groups_pagination_next_token = page_token

return [self.get_group_obj(group_data) for group_data in group_list]

def admin_add_user_to_group(self, username, group_name):
"""
Expand Down Expand Up @@ -934,6 +1042,46 @@ def admin_update_identity_provider(self, pool_id, provider_name, **kwargs):
**kwargs,
)

def list_user_pool_clients(
self,
pool_id: str | None = None,
page_limit: int | None = None,
page_token: str | None = None,
) -> list[dict]:
"""
Returns configuration information of a user pool's clients. If page limit is set
then it will return that many (0 to 60) while setting self._clients_pagination_next_token
to the next token.
:param pool_id: The user pool ID to list clients for (uses self.user_pool_id if None)
:param page_limit: Max results to return from this request (0 to 60)
:param page_token: Used to return the next set of items
:return: List of client dicts of the specified user pool
"""
if pool_id is None:
pool_id = self.user_pool_id

kwargs = {"UserPoolId": pool_id}
if page_limit:
kwargs["MaxResults"] = page_limit
if page_token:
kwargs["NextToken"] = page_token

response = self.client.list_user_pool_clients(**kwargs)
client_list = response.get("UserPoolClients")
page_token = response.get("NextToken")

if page_limit is None:
while page_token:
response = self.client.list_user_pool_clients(
UserPoolId=pool_id, PaginationToken=page_token
)
client_list.extend(response.get("UserPoolClients"))
page_token = response.get("NextToken")
else:
self._clients_pagination_next_token = page_token

return client_list

def describe_user_pool_client(self, pool_id: str, client_id: str):
"""
Returns configuration information of a specified user pool app client
Expand Down
Loading