考虑使用Python编写的AWS lambda,使用boto3作为AWS Redshift服务的客户端。
考虑以下示例:
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')
# List of executed statements (id... sql...)
# Assert Statement Descriptions
使用Moto python依赖来模拟redshift-data
(类似于上面的例子),我如何检索关于执行语句的数据?我指的是之前执行的语句id和在lambda_handler
上调用的sql字符串("INSERT…")我找不到任何东西来获取语句id,以便查询服务并获得语句描述。
是否有一种方法来断言以前执行的sql的文本?
没有官方的方法来检索/验证哪些语句已经在execute_statement
-方法中提供。
但是,您可以使用内部API来验证该数据:
from moto.core import DEFAULT_ACCOUNT_ID
from moto.redshiftdata import redshiftdata_backends
backend = redshiftdata_backends[DEFAULT_ACCOUNT_ID]["us-east-1"] # Use the appropriate account/region
for statement in backend.statements.values():
print(statement.query_string)
请注意,由于它是一个内部API,数据结构可能在版本之间发生变化。