如何在Python中使用boto3 1.17版本参数化Athena查询？

由于boto3 1.17不支持ExecutionParameters参数，你可以通过字符串安全格式化的方式手动替换查询中的占位符，同时需注意规避SQL注入风险。针对你给出的SELECT * FROM table WHERE x in (?) AND y in (?)这类带IN子句的查询，具体实现方案如下：

1. 处理IN子句的列表参数 #

IN子句需要将列表转换为单引号包裹、逗号分隔的字符串后，再替换到查询模板中：

示例代码 #

import boto3

# 初始化Athena客户端
athena_client = boto3.client('athena', region_name='你的区域')

# 原始查询模板
query_template = "SELECT * FROM table WHERE x in ({x_values}) AND y in ({y_values})"

# 待替换的参数列表
x_list = [1, 2, 3]
y_list = ['a', 'b', 'c']

# 格式化参数：字符串类型加单引号，数值类型直接转字符串
def format_sql_param_list(param_list):
    formatted_items = []
    for val in param_list:
        if isinstance(val, str):
            # 转义单引号避免SQL注入
            escaped_val = val.replace("'", "''")
            formatted_items.append(f"'{escaped_val}'")
        else:
            formatted_items.append(str(val))
    return ', '.join(formatted_items)

x_formatted = format_sql_param_list(x_list)
y_formatted = format_sql_param_list(y_list)

# 生成最终查询语句
final_query = query_template.format(x_values=x_formatted, y_values=y_formatted)

# 执行Athena查询
response = athena_client.start_query_execution(
    QueryString=final_query,
    QueryExecutionContext={'Database': '你的数据库名'},
    ResultConfiguration={'OutputLocation': 's3://你的存储桶路径/'}
)

2. 单值参数的替换方案 #

如果是单个值的占位符（比如SELECT * FROM table WHERE x = ?），可以直接做格式化处理：

query_template = "SELECT * FROM table WHERE x = {x_val}"

# 数值类型参数
x_val = 100
final_query = query_template.format(x_val=x_val)

# 字符串类型参数（需转义）
x_val = "user's input"
escaped_x = x_val.replace("'", "''")
final_query = query_template.format(x_val=f"'{escaped_x}'")

关键注意事项 #

• 所有来自不可信来源的字符串参数，必须替换单引号为两个单引号（Athena遵循标准SQL转义规则），避免SQL注入；
• 数值类型参数直接转为字符串即可，但要确保参数确实是数值类型，防止传入恶意字符串。

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