如何用Moto获取Redshift-Data执行的SQL语句及ID？

现有一个基于Python、用boto3调用AWS Redshift Data服务的Lambda函数，代码示例如下：

import boto3
import moto

def lambda_handler(event, context):
    session = boto3.session.Session()

    redshift_data_service = session.client(
        service_name='redshift-data',
        region_name='a_region',
    )

    redshift_data_service.execute_statement(
        sql="insert into XYZ VALUES..."
    )


@moto.redshiftdata
def test_insert_sql_insert():
    lambda_handler(...)
    redshift_data_service = boto3.client("redshift-data", region_name='a_region')
    
    # 想要获取已执行的语句列表（包含ID、SQL内容）
    # 并断言SQL文本是否符合预期

需要解决的核心问题：用Moto模拟Redshift Data服务时，如何获取已执行语句的ID和对应SQL字符串，完成测试断言？

方法1：通过官方API查询（贴近真实调用场景） #

Moto的Redshift Data模拟完全兼容AWS官方API，execute_statement调用会返回包含语句ID的响应。你可以通过以下步骤获取并断言：

1. 修改Lambda函数，捕获并返回执行语句的响应：

def lambda_handler(event, context):
    session = boto3.session.Session()
    redshift_data_service = session.client(
        service_name='redshift-data',
        region_name='a_region',
    )
    # 捕获execute_statement的返回结果
    response = redshift_data_service.execute_statement(
        sql="insert into XYZ VALUES..."
    )
    return response

2. 在测试函数中获取语句ID，查询详情并断言：

@moto.redshiftdata
def test_insert_sql_insert():
    # 调用Lambda并拿到语句响应
    stmt_response = lambda_handler({}, None)
    stmt_id = stmt_response["Id"]
    
    redshift_data_service = boto3.client("redshift-data", region_name='a_region')
    # 查询语句详情
    stmt_details = redshift_data_service.get_statement_details(Id=stmt_id)
    
    # 断言SQL文本是否符合预期
    assert stmt_details["QueryString"] == "insert into XYZ VALUES..."

方法2：直接访问Moto内部存储（高效测试专用） #

Moto会把所有模拟服务的运行状态存在内部后端，你可以直接读取Redshift Data的后端数据，获取所有已执行语句：

@moto.redshiftdata
def test_insert_sql_insert():
    lambda_handler({}, None)
    
    # 获取对应区域的Redshift Data后端实例
    from moto.redshiftdata.models import redshiftdata_backends
    backend = redshiftdata_backends["a_region"]
    
    # 所有已执行语句都存储在backend.statements列表中
    executed_stmts = backend.statements
    # 取最新执行的语句（或按需遍历查找目标语句）
    latest_stmt = executed_stmts[-1]
    
    # 断言语句ID和SQL文本
    assert latest_stmt.id is not None
    assert latest_stmt.sql == "insert into XYZ VALUES..."

注意：方法2属于直接操作Moto内部实现，虽然高效，但如果Moto版本更新可能存在变动，建议优先使用方法1。

内容的提问来源于stack exchange，提问作者Yak O'Poe