diff --git a/README.md b/README.md index c4c41bea..496b6031 100644 --- a/README.md +++ b/README.md @@ -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 @@ -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 @@ -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 diff --git a/pycognito/__init__.py b/pycognito/__init__.py index bd28318a..7026ec11 100644 --- a/pycognito/__init__.py +++ b/pycognito/__init__.py @@ -208,6 +208,10 @@ 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: @@ -215,6 +219,27 @@ def user_pool_url(self): 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 @@ -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"), @@ -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): """ @@ -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 diff --git a/tests.py b/tests.py index 8bc0e5e0..0bbe1d60 100644 --- a/tests.py +++ b/tests.py @@ -496,5 +496,137 @@ def test_srp_requests_http_auth(self, m): self.assertNotEqual(access_token_orig, access_token_new) +@moto.mock_aws +class PaginationTestCase(unittest.TestCase): + invalid_user_pool_id = "us-east-1_123456789" + + def setUp(self) -> None: + + cognito_idp_client = boto3.client("cognito-idp", region_name="us-east-1") + + user_pool = cognito_idp_client.create_user_pool( + PoolName="pycognito-test-pool", + AliasAttributes=[ + "email", + ], + UsernameAttributes=[ + "email", + ], + ) + self.user_pool_id = user_pool["UserPool"]["Id"] + + # create users, groups, and clients to test pagination + for i in range(2): + cognito_idp_client.admin_create_user( + UserPoolId=self.user_pool_id, + Username=f"user{i}@test.com", + TemporaryPassword="Testing123!", + MessageAction="SUPPRESS", + ) + + cognito_idp_client.create_group( + GroupName=f"group-{i}", UserPoolId=self.user_pool_id + ) + + cognito_idp_client.create_user_pool_client( + UserPoolId=self.user_pool_id, + ClientName=f"test-client-{i}", + RefreshTokenValidity=1, + AccessTokenValidity=1, + IdTokenValidity=1, + TokenValidityUnits={ + "AccessToken": "hour", + "IdToken": "hour", + "RefreshToken": "days", + }, + ) + + def test_user_pagination(self): + cognito = Cognito(user_pool_id=self.user_pool_id) + + # retrieve the first user + user_one = cognito.get_users(pool_id=self.user_pool_id, page_limit=1) + self.assertEqual(len(user_one), 1) + + # verify a page token exists for the next request + page_token = cognito.get_users_pagination_token() + self.assertTrue(page_token is not None) + + # retrieve the second user + user_two = cognito.get_users(page_limit=1, page_token=page_token) + self.assertEqual(len(user_two), 1) + + # verify page token doesn't exist since users are exhausted + page_token = cognito.get_users_pagination_token() + self.assertTrue(page_token is None) + + # verify we can retrieve all users via pagination if no limits specified + all_users = cognito.get_users() + self.assertEqual(len(all_users), 2) + + # test that a different user pool id can be specified for the request + cognito.user_pool_id = self.invalid_user_pool_id + users = cognito.get_users(pool_id=self.user_pool_id) + self.assertEqual(len(users), 2) + + def test_group_pagination(self): + cognito = Cognito(user_pool_id=self.user_pool_id) + + # retrieve the first group + group_one = cognito.get_groups(pool_id=self.user_pool_id, page_limit=1) + self.assertEqual(len(group_one), 1) + + # verify a page token exists for the next request + page_token = cognito.get_groups_pagination_token() + self.assertTrue(page_token is not None) + + # retrieve the second group + group_two = cognito.get_groups(page_limit=1, page_token=page_token) + self.assertEqual(len(group_two), 1) + + # verify page token doesn't exist since groups are exhausted + page_token = cognito.get_groups_pagination_token() + self.assertTrue(page_token is None) + + # verify we can retrieve all groups via pagination if no limits specified + all_groups = cognito.get_groups() + self.assertEqual(len(all_groups), 2) + + # test that a different user pool id can be specified for the request + cognito.user_pool_id = self.invalid_user_pool_id + groups = cognito.get_groups(pool_id=self.user_pool_id) + self.assertEqual(len(groups), 2) + + def test_client_pagination(self): + cognito = Cognito(user_pool_id=self.user_pool_id) + + # retrieve the first client + client_one = cognito.list_user_pool_clients( + pool_id=self.user_pool_id, page_limit=1 + ) + self.assertEqual(len(client_one), 1) + + # verify a page token exists for the next request + page_token = cognito.get_clients_pagination_token() + self.assertTrue(page_token is not None) + + # retrieve the second client + client_two = cognito.list_user_pool_clients(page_limit=1, page_token=page_token) + self.assertEqual(len(client_two), 1) + + # verify page token doesn't exist since clients are exhausted + page_token = cognito.get_clients_pagination_token() + self.assertTrue(page_token is None) + + # verify we can retrieve all clients via pagination if no limits specified + all_clients = cognito.list_user_pool_clients() + self.assertEqual(len(all_clients), 2) + + # test that a different user pool id can be specified for the request + cognito.user_pool_id = self.invalid_user_pool_id + clients = cognito.list_user_pool_clients(pool_id=self.user_pool_id) + self.assertEqual(len(clients), 2) + + if __name__ == "__main__": unittest.main()