AWS Lambda boto3でCognitoのユーザーを削除する

はじめに

AWS LambdaでCognitoのユーザーを削除します。
ランタイムはpythonです。
環境変数COGNITO_MAX_RESULTSには、50を指定します。
ユーザープールの数が50以下である事を想定しています。
Cognitoからユーザーを削除する場合、ループ処理となり、たまにコケることがあるため、リトライ処理を追加しています。

Lambdaのスクリプト

Lambda

import os
import boto3
import time
import logging
from typing import List, Dict, Any

# Logger設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Cognitoクライアントの初期化
cognito = boto3.client('cognito-idp', region_name='ap-northeast-1')

def handler(event: Dict[str, Any], context: Any) -> List[Dict[str, Any]]:
    """Lambdaハンドラー

    Args:
        event (Dict[str, Any]): イベントデータ
        context (Any): Lambda実行コンテキスト

    Returns:
        List[Dict[str, Any]]: ユーザー削除結果のリスト
    """
    user_pool_name = 'ユーザープール名'
    user_ids = ['ユーザーID']

    logger.info("Starting user deletion process for pool: %s", user_pool_name)
    response = delete_cognito_users(user_pool_name, user_ids)
    return response

def get_user_pool_id(user_pool_name: str) -> str:
    """指定されたユーザープール名からユーザープールIDを取得

    Args:
        user_pool_name (str): Cognitoユーザープール名

    Returns:
        str: CognitoユーザープールID

    Raises:
        ValueError: ユーザープールが見つからない場合
    """
    max_results = int(os.getenv('COGNITO_MAX_RESULTS', 50))
    logger.debug("Fetching user pool list with max results: %d", max_results)

    user_pools = cognito.list_user_pools(MaxResults=max_results)

    for user_pool in user_pools.get('UserPools', []):
        if user_pool['Name'] == user_pool_name:
            logger.info("Found user pool ID for pool: %s", user_pool_name)
            return user_pool['Id']
    
    logger.error("User pool '%s' not found.", user_pool_name)
    raise ValueError(f"User pool '{user_pool_name}' not found.")

def delete_cognito_users(user_pool_name: str, user_ids: List[str]) -> List[Dict[str, Any]]:
    """Cognitoから指定されたユーザーを削除

    Args:
        user_pool_name (str): Cognitoユーザープール名
        user_ids (List[str]): 削除対象のユーザーIDリスト

    Returns:
        List[Dict[str, Any]]: 各ユーザーの削除結果のリスト
    """
    user_pool_id = get_user_pool_id(user_pool_name)
    results = []

    for user_id in user_ids:
        while True:
            try:
                response = cognito.admin_delete_user(
                    UserPoolId=user_pool_id,
                    Username=user_id
                )
                logger.info("Successfully deleted user: %s", user_id)
                results.append({'user_id': user_id, 'status': 'deleted'})
                break
            except cognito.exceptions.TooManyRequestsException:
                logger.warning("Rate limit exceeded for user %s. Retrying in 5 seconds...", user_id)
                time.sleep(5)  # レート制限のため再試行
            except cognito.exceptions.UserNotFoundException:
                logger.warning("User not found: %s", user_id)
                results.append({'user_id': user_id, 'status': 'not_found'})
                break
            except Exception as e:
                logger.error("Failed to delete user %s: %s", user_id, str(e))
                results.append({'user_id': user_id, 'status': 'error', 'error': str(e)})
                break
    return results