[Boto3]ListUsersAPIでCognitoのユーザーを検索する

ドキュメントでは、以下のようにしてユーザーを取得できるという風に書かれています。

response = client.list_users(
    UserPoolId='string',
    AttributesToGet=[
        'string',
    ],
    Limit=123,
    PaginationToken='string',
    Filter='string'
)

from:https://boto3.readthedocs.io/en/latest/reference/services/cognito-idp.html#CognitoIdentityProvider.Client.list_users

##ユーザーを60件取得

import boto3
client = boto3.client('cognito-idp')
response = client.list_users(
    UserPoolId='YOUR_USER_POOL_ID'
)

ユーザープールIDだけ指定した場合、全ユーザー先頭60件が返ってきます。
「Limitを指定しないと全件取得する」と思っていましたが、先頭60件のみの様子です。

##メールアドレスで検索する
Filterにemail = "hoge"と書くことで、該当するEmailアドレスを持つユーザーのみを検索することができます。

import boto3
client = boto3.client('cognito-idp')
response = client.list_users(
    UserPoolId='YOUR_USER_POOL_ID',
    Filter="email = \"hide\"",
)

この例の場合、hideという文字が含まれるメールアドレスを使用するユーザーを全件取得します。

=の代わりに^=とすることで、「hideから始まるメールアドレス」のような検索もできます。

Tips:カスタム属性はFilterで検索できない

標準属性のみを検索できます。カスタム属性は検索できません。これはインデックスが付けられた属性のみが検索可能なためで、カスタム属性にインデックスを作成することはできません。
http://docs.aws.amazon.com/ja_jp/cognito/latest/developerguide/how-to-manage-user-accounts.html#cognito-user-pools-searching-for-users-using-listusers-api

とあり、カスタム属性を使用したユーザー検索は現状対応していない様子です。

検索させたい属性については、DynamoDBやAES(Amazon Elasticsearch Service)などに格納する方が良さそうです。

##先頭10ユーザーを取得する

Limitクエリに数字を指定することで、取得件数を絞り込めます。

import boto3
client = boto3.client('cognito-idp')
response = client.list_users(
    UserPoolId='YOUR_USER_POOL_ID',
    Limit=10,
)

Tips: Limitの上限は60

61件以上のデータを一気に取得しようとして、Limitを61以上の値にするとエラーになります。

ClientError: An error occurred (InvalidParameterException) when calling the ListUsers operation: 1 validation error detected: Value '500' at 'limit' failed to satisfy constraint: Member must have value less than or equal to 60

以下のような感じで再帰呼び出しするのがいいかなと思います。

def getUser():
    client = boto3.client('cognito-idp')
    response = self.client.list_users(
        UserPoolId="YOUR_USERPOOL_ID"
    )
    return response

def getUsersWithPagenation(PaginationToken):
    response = self.client.list_users(
        UserPoolId="YOUR_USERPOOL_ID",
        PaginationToken=PaginationToken
    )
    return response

def workerNode(users):
    for user in users["Users"]:
    	# Do Something
    if 'PaginationToken' in users:
        newUsers = getUsersWithPagenation(users["PaginationToken"])
        workerNode(newUsers)

users = getUser()
workerNode(users)