You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

在AWS Lambda中获取Cognito用户池用户时遇JSON序列化错误

解决Lambda返回Cognito用户列表时的JSON序列化错误

问题根源是Cognito list_users 接口返回的响应中包含datetime类型字段(比如用户创建时间UserCreateDate、最后修改时间UserLastModifiedDate),而Lambda要求返回值必须是JSON可序列化的对象,datetime类型无法直接被JSON序列化,因此触发报错。

以下是两种可行的解决方法:

方法一:手动转换datetime字段为字符串

遍历响应中的用户数据,将datetime类型字段转换为ISO格式的字符串:

import boto3
from datetime import datetime

client = boto3.client('cognito-idp')

def lambda_handler(event, context):
    response = client.list_users(UserPoolId='us-east-1_TIzqd0Fik')
    
    # 转换用户级别的datetime字段
    for user in response['Users']:
        if 'UserCreateDate' in user:
            user['UserCreateDate'] = user['UserCreateDate'].isoformat()
        if 'UserLastModifiedDate' in user:
            user['UserLastModifiedDate'] = user['UserLastModifiedDate'].isoformat()
    
    return response

方法二:用自定义序列化函数批量处理

通过JSON序列化工具的default参数,自动处理所有datetime类型:

import boto3
import json
from datetime import datetime

client = boto3.client('cognito-idp')

def datetime_serializer(obj):
    # 把datetime转换为ISO格式字符串
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError("Type not serializable")

def lambda_handler(event, context):
    response = client.list_users(UserPoolId='us-east-1_TIzqd0Fik')
    # 序列化后转回字典,确保所有不可序列化类型都被处理
    serialized_response = json.loads(json.dumps(response, default=datetime_serializer))
    return serialized_response

补充:处理分页获取所有用户

list_users默认最多返回60个用户,如果用户池用户数量较多,需要通过分页获取全部数据:

import boto3
import json
from datetime import datetime

client = boto3.client('cognito-idp')

def datetime_serializer(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError("Type not serializable")

def lambda_handler(event, context):
    all_users = []
    pagination_token = None
    
    while True:
        request_params = {'UserPoolId': 'us-east-1_TIzqd0Fik'}
        if pagination_token:
            request_params['PaginationToken'] = pagination_token
        
        response = client.list_users(**request_params)
        all_users.extend(response['Users'])
        
        # 检查是否还有下一页数据
        pagination_token = response.get('PaginationToken')
        if not pagination_token:
            break
    
    # 序列化所有用户数据并返回
    serialized_users = json.loads(json.dumps(all_users, default=datetime_serializer))
    return {'Users': serialized_users}

内容的提问来源于stack exchange,提问作者Showri Kranthi

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.11 15:25:19