Python中如何使用Boto3正确处理AWS Cognito客户端异常

你当前的实现存在明显问题，不是处理Cognito异常的正确方案，核心问题如下：

• 异常捕获逻辑存在变量未定义风险：你将boto3.client初始化放在try代码块内部，如果初始化过程报错（比如区域配置错误、AWS凭证缺失/无效、网络不通等），client变量根本不会被创建，后续except client.exceptions.XXX的写法会直接抛出NameError，导致整个请求直接500，根本达不到异常捕获效果。
• 异常覆盖范围不足：当前只捕获了用户名重复的异常，Cognito的sign_up接口还会抛出十多种业务异常，比如密码不符合复杂度要求、邮箱格式错误、客户端ID无效等，这些异常都没有被处理，会直接向上抛出导致服务报错。
• 硬编码的用户确认校验逻辑不合理：默认情况下Cognito用户注册后UserConfirmed就是False，需要走邮箱/手机号验证流程才会变为True，你直接判断这个字段为False就返回错误，会导致所有新注册用户都无法正常走后续验证流程，属于业务逻辑错误。

优化后的实现方案 #

1. 把Cognito客户端初始化逻辑移出异常捕获块，或者单独对初始化阶段的异常做捕获，避免服务启动后运行时才发现配置问题
2. 直接通过botocore的通用异常类捕获AWS服务错误，不需要依赖实例化后的client变量获取异常类
3. 覆盖所有常见的业务异常，同时兜底处理未预知的其他异常
4. 修正UserConfirmed字段的判断逻辑，按照业务要求选择引导用户验证或者走自动确认流程

修正后的代码示例

import boto3
from botocore.exceptions import ClientError
import os
from rest_framework import status
from rest_framework.response import Response

# 建议把客户端初始化放在全局位置，避免每次请求都重复创建
try:
    COGNITO_CLIENT = boto3.client(
        "cognito-idp", 
        region_name=settings.COGNITO_AWS_REGION
    )
except Exception as e:
    # 初始化失败直接抛出启动错误，提前暴露配置问题
    raise RuntimeError(f"Cognito客户端初始化失败: {str(e)}") from e

def post(self, request):
    # 提前校验必填参数，避免不必要的Cognito调用
    required_fields = ["username", "password", "email"]
    for field in required_fields:
        if field not in request.data:
            return Response(
                dict(error=f"缺少必填参数{field}", user_msg="参数不完整，请检查后重试"),
                status=status.HTTP_400_BAD_REQUEST
            )
    
    client_id = os.getenv(settings.COGNITO_CLIENT_ID)
    if not client_id:
        return Response(
            dict(error="Cognito客户端ID配置缺失", user_msg="服务配置异常，请联系管理员"),
            status=status.HTTP_500_INTERNAL_SERVER_ERROR
        )

    try:
        response = COGNITO_CLIENT.sign_up(
            ClientId=client_id,
            Username=request.data['username'],
            Password=request.data['password'],
            UserAttributes=[{"Name": "email", "Value": request.data['email']}],
        )
        request.data['uuid'] = response['UserSub']

        # 可根据业务逻辑调整，默认新注册用户都是未确认状态，不需要返回错误
        if not response['UserConfirmed']:
            return Response(
                dict(
                    uuid=response['UserSub'],
                    user_msg="注册成功，请前往邮箱查收验证邮件完成账号激活"
                ),
                status=status.HTTP_200_OK
            )
        
        return Response(
            dict(uuid=response['UserSub'], user_msg="注册成功"),
            status=status.HTTP_200_OK
        )

    except ClientError as e:
        error_code = e.response["Error"]["Code"]
        error_msg_map = {
            "UsernameExistsException": "该用户名已被注册",
            "InvalidPasswordException": "密码不符合复杂度要求，请包含大小写字母、数字和特殊字符，长度不少于8位",
            "InvalidParameterException": "参数格式错误，请检查邮箱、用户名格式是否正确",
            "UserLambdaValidationException": "注册信息未通过校验，请检查输入内容",
            "NotAuthorizedException": "当前客户端无注册权限，请联系管理员",
            "LimitExceededException": "请求过于频繁，请稍后再试",
        }
        user_msg = error_msg_map.get(error_code, "注册失败，请稍后再试")
        return Response(
            dict(error=str(e), user_msg=user_msg),
            status=status.HTTP_400_BAD_REQUEST if error_code in error_msg_map else status.HTTP_500_INTERNAL_SERVER_ERROR
        )
    except Exception as e:
        # 兜底捕获其他未知异常
        return Response(
            dict(error=str(e), user_msg="服务异常，请稍后再试"),
            status=status.HTTP_500_INTERNAL_SERVER_ERROR
        )

内容的提问来源于stack exchange，提问作者user12314098