SQLAlchemy execute()参数问题：传入元组时执行失败

错误根源 #

1. 参数格式不匹配：SQLAlchemy的connection.execute()方法处理位置占位符（?）时，若传入单条数据的元组，需要将其包装为嵌套的可迭代结构（比如列表套元组）。直接传单个元组会被方法误认为是多个独立参数，而非一组对应占位符的参数集合，这就触发了List argument must consist only of tuples or dictionaries错误。
2. 默认值处理逻辑冲突：你的函数将params_tuple为None时默认设为{}（空字典），但空字典是给命名占位符（如:position_date）使用的，和位置占位符的参数格式不兼容，可能在无参数场景埋下隐患。

解决办法 #

方法一：调整参数传入格式 #

调用函数时，把单条元组包装成列表：

execute_query(
    "INSERT INTO x.aggregation_dates (position_date, account_id, horizon) VALUES (?, ?, ?)",
    engine,
    params_tuple=[('2025-01-31', 12, 22)]
)

同时修改函数的默认值处理，匹配位置参数的场景：

def execute_query(sql_query, engine, params=None):
    params = params if params else ()
    
    with engine.connect() as connection:
        result = connection.execute(text(sql_query), params)
        connection.commit()

方法二：在函数内自动兼容单条元组 #

修改函数逻辑，自动识别单条元组并包装，避免调用方调整：

def execute_query(sql_query, engine, params=None):
    params = params if params else ()
    # 若传入的是单条数据元组，自动包装为列表
    if isinstance(params, tuple) and params and not isinstance(params[0], (tuple, dict)):
        params = [params]
    
    with engine.connect() as connection:
        result = connection.execute(text(sql_query), params)
        connection.commit()

此时调用时直接传原元组即可：

execute_query(
    "INSERT INTO x.aggregation_dates (position_date, account_id, horizon) VALUES (?, ?, ?)",
    engine,
    params_tuple=('2025-01-31', 12, 22)
)

补充说明 #

SQLAlchemy的execute方法参数规则：

• 使用位置占位符（?）时，参数可以是：包含多个元组的列表（批量插入）、单个元组（部分版本支持，但更稳妥的是包装为列表）；
• 使用命名占位符（如:position_date）时，参数需为字典，此时你的原默认值{}才适用。

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