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 list_user_pool_clients and fine grained pagination support #269

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Add list_user_pool_clients and fine grained pagination support #269

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
View file
Open in desktop
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
148 changes: 133 additions & 15 deletions pycognito/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,13 +208,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 @@ -577,25 +602,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 +851,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 +1012,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