如何在Step Functions内的Lambda中获取上次成功执行时间替换固定1小时偏移

需求可行性结论 #

该需求完全可实现，针对你运行在Step Function内部、通过Terraform部署的场景，现有逻辑可直接改造适配，你参考的代码片段逻辑可以复用。

实现方案说明 #

你提到的get-execution-history接口不需要使用，该接口主要用于查询单次Step Function执行内部的事件流转记录，你需要获取的是状态机最近一次成功执行的时间，使用list_executions接口更适配需求，具体实现步骤如下：

1. 权限配置（Terraform侧） #

首先给Lambda的执行角色授予调用Step Function接口的权限，同时将Step Function的ARN通过环境变量注入Lambda方便代码读取，参考Terraform配置如下：

# 给Lambda角色添加SFN接口调用权限
resource "aws_iam_role_policy" "sfn_list_executions" {
  name   = "allow-list-sfn-executions"
  role   = aws_iam_role.lambda_exec_role.id # 替换为你的Lambda执行角色资源名
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action   = "states:ListExecutions"
        Effect   = "Allow"
        Resource = aws_sfn_state_machine.your_sfn.arn # 替换为你的Step Function资源名
      }
    ]
  })
}

# 注入SFN ARN到Lambda环境变量
resource "aws_lambda_function" "api_pull_lambda" {
  # 保留你原有Lambda配置不变
  environment {
    variables = {
      STATE_MACHINE_ARN = aws_sfn_state_machine.your_sfn.arn
    }
  }
}

2. Lambda代码改造 #

你参考的代码只需微调即可落地，注意要过滤掉当前正在运行的本次执行记录，参考代码如下：

import boto3
import os
from datetime import datetime, timedelta

sfn_client = boto3.client('stepfunctions')

# 原有代码逻辑...
from_time = ""
# 拉取状态机成功执行的历史记录，按启动时间倒序
response = sfn_client.list_executions(
    stateMachineArn=os.getenv('STATE_MACHINE_ARN'),
    statusFilter='SUCCEEDED',
    maxResults=2
)

if response["executions"] and len(response["executions"]) >= 2:
    # 索引0为本次正在运行的执行，索引1为上一次成功执行的记录
    last_success_start_time = response["executions"][1]["startDate"]
    week_ago = datetime.now() - timedelta(days=7)
    week_ago = week_ago.replace(tzinfo=last_success_start_time.tzinfo)
    # 保留原有兜底逻辑：上次执行超过7天则默认拉取1小时数据
    if last_success_start_time > week_ago:
        from_time = get_utc_time(last_success_start_time)
    else:
        from_time = get_utc_time(datetime.now() - timedelta(hours=1))
else:
    # 首次执行无历史记录，兜底拉取1小时数据
    from_time = get_utc_time(datetime.now() - timedelta(hours=1))

logger.info(f"Last successful execution time in UTC: {from_time}")
# 后续拉取API逻辑不变...

可选优化方案 #

如果你不想依赖Step Function接口调用，也可以选择用DynamoDB存储上次成功执行的时间戳：

• 新建一个单条记录的DynamoDB表，存放上次执行成功的时间
• Lambda执行成功后更新该表的时间戳
• 每次启动时先读取该时间戳作为拉取起始时间
  该方案性能更高，也不受Step Function执行记录保留周期的限制。

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