Python(AWS Lambda)函数传参时Amazon Pinpoint提示Segment不存在的问题？

• 核心问题：create_import_job是Pinpoint提供的异步接口，调用后仅返回任务的初始状态，此时关联的Segment还在后台导入CSV端点数据、完成资源构建，并未生成可被Campaign关联的有效Segment，立刻调用create_campaign自然触发Segment不存在的报错。
• 逻辑缺陷：check_import_jobs的去重逻辑完全失效，代码未读取导入任务返回的真实S3Url，反而硬编码将full_path塞入对比列表，无法正确识别已存在的导入任务，可能触发重复创建。

1. 新增导入任务状态轮询逻辑 #

拿到create_import_job返回的任务ID后，循环查询任务状态，直到任务完成、Segment完全生成后，再返回有效的SegmentID用于后续Campaign创建。
修正后的create_import_job实现：

def create_import_job(s3_url, full_path):
    client = boto3.client('pinpoint')
    segment_name = s3_url.split('/')[4]
    try:
        response = client.create_import_job(
            ApplicationId=projectId,
            ImportJobRequest={
                'DefineSegment': True,
                'Format': 'CSV',
                'RegisterEndpoints': True,
                'RoleArn': importRoleArn,
                'S3Url': s3_url,
                'SegmentName': segment_name
            }
        )
    except ClientError as e:
        print(e.response['Error']['Message'])
        return None
    
    import_job_id = response['ImportJobResponse']['Id']
    segment_id = response['ImportJobResponse']['Definition']['SegmentId']
    print(f"Import job {import_job_id} initial status: {response['ImportJobResponse']['JobStatus']}")

    # 轮询任务状态，最长等待15分钟（匹配Lambda最大超时时间）
    max_wait = 15 * 60
    wait_interval = 10
    waited = 0
    while waited < max_wait:
        job_resp = client.get_import_job(
            ApplicationId=projectId,
            JobId=import_job_id
        )
        current_status = job_resp['ImportJobResponse']['JobStatus']
        if current_status == 'COMPLETED':
            print(f"Import job {import_job_id} completed")
            return segment_id
        elif current_status in ('FAILED', 'CANCELLED'):
            print(f"Import job {import_job_id} ended with status: {current_status}")
            return None
        time.sleep(wait_interval)
        waited += wait_interval
    
    print(f"Import job {import_job_id} timed out")
    return None

2. 修正导入任务去重逻辑 #

调整check_import_jobs，正确读取已存在导入任务的真实S3Url做对比：

def check_import_jobs(bucket, folder, s3_url, full_path):
    url_list = []
    print(s3_url)
    client = boto3.client('pinpoint')
    try:
        client_response = client.get_import_jobs(
            ApplicationId=projectId
        )
    except ClientError as e:
        print(e.response['Error']['Message'])
        return False
    else:
        segment_response = client_response['ImportJobsResponse']['Item']
        for item in segment_response:
            # 读取每个导入任务实际配置的S3Url
            url_list.append(item['ImportJobRequest']['S3Url'])
    return s3_url in url_list

3. 调整主流程分支判断 #

仅当导入任务完成、拿到有效SegmentID时，才触发Campaign创建：

if not (check_import_jobs(bucket, folder, s3_url, full_path)):
    segmentID = create_import_job(s3_url, full_path)
    if segmentID:
        create_campaign(segmentID)

• 将Lambda函数的超时时间调整为15分钟，默认3秒的超时远不足以等待导入任务完成。
• 轮询间隔、最长等待时间可根据实际CSV文件大小调整，数据量越大适当拉长等待上限。
• 禁止用固定时长的time.sleep替代状态轮询，固定延时要么浪费Lambda运行成本，要么等待时间不足导致Segment未生成。

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